# retoor import pytest from devplacepy.content import can_manage_workspace, can_open_workspace from devplacepy.database import get_table, init_db, set_setting from devplacepy.services.containers import activity, store from devplacepy.services.containers.workspace import ( flags, naming, provision, quota, tunnels, ) from devplacepy.services.containers.workspace.provision import WorkspaceError from tests.conftest import run_async OWNER = "user-owner" OTHER = "user-other" @pytest.fixture(autouse=True) def _workspace_db(): init_db() set_setting("workspace_enabled", "1") yield for table in ("instances", "tunnels", "workspace_flags", "workspace_quota_rules"): get_table(table).delete() def _project(uid: str = "proj-ws") -> dict: return {"uid": uid, "slug": "demo", "title": "Demo", "user_uid": OWNER} def _instance(**overrides) -> dict: row = { "project_uid": "proj-ws", "name": "ws-demo", "status": "running", "desired_state": "running", "is_workspace": 1, "workspace_owner_uid": OWNER, "tunnel_name": naming.generate(), "ports_json": '[{"host": 20500, "container": 8080, "proto": "tcp"}]', } row.update(overrides) return store.create_instance(row) def test_open_workspace_requires_enabled_setting(): set_setting("workspace_enabled", "0") assert can_open_workspace(_project(), {"uid": OWNER, "role": "Member"}) is False set_setting("workspace_enabled", "1") assert can_open_workspace(_project(), {"uid": OWNER, "role": "Member"}) is True def test_guest_can_never_open_workspace(): assert can_open_workspace(_project(), None) is False assert can_open_workspace(_project(), {}) is False def test_non_owner_member_cannot_open_workspace(): assert can_open_workspace(_project(), {"uid": OTHER, "role": "Member"}) is False def test_admin_can_open_any_project_workspace(): assert can_open_workspace(_project(), {"uid": OTHER, "role": "Admin"}) is True def test_owner_manages_own_workspace_without_admin(): instance = _instance() assert can_manage_workspace(instance, _project(), {"uid": OWNER, "role": "Member"}) assert not can_manage_workspace( instance, _project(), {"uid": OTHER, "role": "Member"} ) def test_create_or_resume_is_idempotent(): project = _project() user = {"uid": OWNER, "username": "owner"} first = run_async(provision.ensure(project, user)) second = run_async(provision.ensure(project, user)) assert first["uid"] == second["uid"] assert provision.count_for_owner(OWNER) == 1 def test_workspace_quota_blocks_beyond_limit(): set_setting("workspace_max_per_user", "1") user = {"uid": OWNER, "username": "owner"} run_async(provision.ensure(_project("p-a"), user)) with pytest.raises(WorkspaceError): run_async(provision.ensure(_project("p-b"), user)) set_setting("workspace_max_per_user", "2") def test_tunnel_revives_rather_than_duplicates(): instance = _instance() first = tunnels.create(instance, "web", 8080, OWNER) tunnels.soft_delete(first["uid"], OWNER) assert tunnels.count_for_instance(instance["uid"]) == 0 revived = tunnels.create(instance, "web", 8080, OWNER) assert revived["uid"] == first["uid"] assert tunnels.count_for_instance(instance["uid"]) == 1 def test_tunnel_hostname_is_a_single_dns_label(): instance = _instance() row = tunnels.create(instance, "web", 3000, OWNER) host = row["hostname"] suffix = "." + naming.domain() assert host.endswith(suffix) label = host[: -len(suffix)] assert "." not in label assert naming.is_valid_label(label) def test_suspend_stops_tunnels_without_deleting_them(): instance = _instance() tunnels.create(instance, "web", 8080, OWNER) provision.suspend(instance, "admin-uid", "abuse") rows = tunnels.list_for_instance(instance["uid"]) assert rows and rows[0]["status"] == "suspended" refreshed = store.get_instance(instance["uid"]) assert refreshed["suspended_at"] provision.unsuspend(refreshed) assert tunnels.list_for_instance(instance["uid"])[0]["status"] == "pending" def test_suspended_workspace_cannot_resume(): instance = _instance() provision.suspend(instance, "admin-uid", "abuse") with pytest.raises(WorkspaceError): provision.resume(store.get_instance(instance["uid"])) def test_flags_are_idempotent_while_open(): instance = _instance() first = flags.raise_flag(instance, flags.KIND_EGRESS, "warn", "a", 1.0, 0.5) second = flags.raise_flag(instance, flags.KIND_EGRESS, "warn", "b", 2.0, 0.5) assert first["uid"] == second["uid"] assert len(flags.list_flags(instance_uid=instance["uid"])) == 1 flags.clear_flag(instance["uid"], flags.KIND_EGRESS, "admin") assert flags.list_flags(instance_uid=instance["uid"]) == [] def test_activity_accumulates_egress_and_requests(): instance = _instance() activity.forget(instance["uid"]) for _ in range(3): activity.touch(instance["uid"], egress_bytes=100) activity.flush(instance["uid"]) row = store.get_instance(instance["uid"]) assert row["egress_bytes"] == 300 assert row["request_count"] == 3 assert row["last_active_at"] def test_quota_rule_overrides_only_its_owner(): get_table("workspace_quota_rules").insert( { "uid": "rule-x", "owner_kind": "user", "owner_id": OWNER, "label": "power", "max_workspaces": 7, "deleted_at": None, "deleted_by": None, } ) assert quota.resolve(OWNER).max_workspaces == 7 assert quota.resolve(OTHER).max_workspaces == quota.resolve().max_workspaces def test_workspace_env_contract_is_complete(): from devplacepy.services.containers import api instance = _instance() env = api.workspace_env(instance, "https://example.test") for key in ( "DEVPLACE_WORKSPACE", "DEVPLACE_WORKSPACE_UID", "DEVPLACE_TUNNEL_NAME", "DEVPLACE_TUNNEL_DOMAIN", "DEVPLACE_TUNNEL_MANIFEST", "VSCODE_PROXY_URI", "DEVPLACE_EDITOR_PORT", "DEVPLACE_QUOTA_DISK_MB", "DEVPLACE_RETENTION_DAYS", ): assert key in env, key assert all(isinstance(value, str) for value in env.values()) assert "{{port}}" in env["VSCODE_PROXY_URI"] def test_non_workspace_instance_gets_no_workspace_env(): from devplacepy.services.containers import api env = api.workspace_env({"is_workspace": 0}, "https://example.test") assert env == {"DEVPLACE_WORKSPACE": ""} def test_devii_workspace_tools_are_role_gated(): from devplacepy.services.devii.registry import CATALOG names = {a.name for a in CATALOG.actions if a.handler == "workspace"} admin_only = { a.name for a in CATALOG.actions if a.handler == "workspace" and a.requires_admin } guest = {s["function"]["name"] for s in CATALOG.tool_schemas_for(False, False)} member = {s["function"]["name"] for s in CATALOG.tool_schemas_for(True, False)} admin = {s["function"]["name"] for s in CATALOG.tool_schemas_for(True, True)} assert not (names & guest) assert not ((names & member) & admin_only) assert names <= admin def test_every_confirm_gated_tool_declares_a_confirm_param(): from devplacepy.services.devii.actions.dispatcher import CONFIRM_REQUIRED from devplacepy.services.devii.registry import CATALOG for action in CATALOG.actions: if action.handler != "workspace" or action.name not in CONFIRM_REQUIRED: continue assert any(p.name == "confirm" for p in action.params), action.name def test_tunnel_host_detection_never_matches_the_site(): assert naming.is_tunnel_host("abc." + naming.domain()) assert not naming.is_tunnel_host("pravda.education") assert not naming.is_tunnel_host("")