113 lines
4.7 KiB
Python
113 lines
4.7 KiB
Python
"""Tests for PluginRegistry user-scoped session isolation (remote-hosted mode)."""
|
|
|
|
import pytest
|
|
|
|
from core.config import config
|
|
from transport.plugin_registry import PluginRegistry
|
|
|
|
|
|
class TestRegistryUserIsolation:
|
|
@pytest.mark.asyncio
|
|
async def test_register_with_user_id_stores_composite_key(self):
|
|
registry = PluginRegistry()
|
|
session = await registry.register(
|
|
"sess-1", "MyProject", "hash1", "2022.3", user_id="user-A"
|
|
)
|
|
assert session.user_id == "user-A"
|
|
assert ("user-A", "hash1") in registry._user_hash_to_session
|
|
assert registry._user_hash_to_session[("user-A", "hash1")] == "sess-1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_id_by_hash(self):
|
|
registry = PluginRegistry()
|
|
await registry.register("sess-1", "Proj", "h1", "2022", user_id="uA")
|
|
|
|
found = await registry.get_session_id_by_hash("h1", "uA")
|
|
assert found == "sess-1"
|
|
|
|
# Different user, same hash -> not found
|
|
not_found = await registry.get_session_id_by_hash("h1", "uB")
|
|
assert not_found is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_user_isolation_same_hash(self):
|
|
"""Two users registering with the same project_hash get independent sessions."""
|
|
registry = PluginRegistry()
|
|
sess_a = await registry.register("sA", "Proj", "hash1", "2022", user_id="userA")
|
|
sess_b = await registry.register("sB", "Proj", "hash1", "2022", user_id="userB")
|
|
|
|
assert sess_a.session_id == "sA"
|
|
assert sess_b.session_id == "sB"
|
|
|
|
# Each user resolves to their own session
|
|
assert await registry.get_session_id_by_hash("hash1", "userA") == "sA"
|
|
assert await registry.get_session_id_by_hash("hash1", "userB") == "sB"
|
|
|
|
# Both sessions exist
|
|
all_sessions = await registry.list_sessions()
|
|
assert len(all_sessions) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions_filtered_by_user(self):
|
|
registry = PluginRegistry()
|
|
await registry.register("s1", "ProjA", "hA", "2022", user_id="userA")
|
|
await registry.register("s2", "ProjB", "hB", "2022", user_id="userB")
|
|
await registry.register("s3", "ProjC", "hC", "2022", user_id="userA")
|
|
|
|
user_a_sessions = await registry.list_sessions(user_id="userA")
|
|
assert len(user_a_sessions) == 2
|
|
assert "s1" in user_a_sessions
|
|
assert "s3" in user_a_sessions
|
|
|
|
user_b_sessions = await registry.list_sessions(user_id="userB")
|
|
assert len(user_b_sessions) == 1
|
|
assert "s2" in user_b_sessions
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions_no_filter_returns_all_in_local_mode(self):
|
|
"""In local mode (not remote-hosted), list_sessions(user_id=None) returns all."""
|
|
registry = PluginRegistry()
|
|
await registry.register("s1", "P1", "h1", "2022", user_id="uA")
|
|
await registry.register("s2", "P2", "h2", "2022", user_id="uB")
|
|
await registry.register("s3", "P3", "h3", "2022") # local mode, no user_id
|
|
|
|
all_sessions = await registry.list_sessions(user_id=None)
|
|
assert len(all_sessions) == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions_no_filter_raises_in_remote_hosted(self, monkeypatch):
|
|
"""In remote-hosted mode, list_sessions(user_id=None) raises ValueError."""
|
|
monkeypatch.setattr(config, "http_remote_hosted", True)
|
|
|
|
registry = PluginRegistry()
|
|
await registry.register("s1", "P1", "h1", "2022", user_id="uA")
|
|
|
|
with pytest.raises(ValueError, match="requires user_id"):
|
|
await registry.list_sessions(user_id=None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unregister_cleans_user_scoped_mapping(self):
|
|
registry = PluginRegistry()
|
|
await registry.register("s1", "Proj", "h1", "2022", user_id="uA")
|
|
assert ("uA", "h1") in registry._user_hash_to_session
|
|
|
|
await registry.unregister("s1")
|
|
|
|
assert ("uA", "h1") not in registry._user_hash_to_session
|
|
assert "s1" not in (await registry.list_sessions())
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconnect_replaces_previous_session(self):
|
|
"""Same (user_id, hash) re-registered evicts old session, stores new one."""
|
|
registry = PluginRegistry()
|
|
await registry.register("old-sess", "Proj", "h1", "2022", user_id="uA")
|
|
assert await registry.get_session_id_by_hash("h1", "uA") == "old-sess"
|
|
|
|
await registry.register("new-sess", "Proj", "h1", "2022", user_id="uA")
|
|
assert await registry.get_session_id_by_hash("h1", "uA") == "new-sess"
|
|
|
|
# Old session should be evicted
|
|
all_sessions = await registry.list_sessions()
|
|
assert "old-sess" not in all_sessions
|
|
assert "new-sess" in all_sessions
|