from __future__ import annotations import json import sys from pathlib import Path import pytest from gyxx_flow.adapters import ( BrowserCookieStore, RuntimeIntegrationCatalog, RuntimeIntegrationError, RuntimeServicePolicy, ) from gyxx_flow.adapters.bootstrap import _rewrite_browser_arguments from gyxx_flow.adapters.native import ModuleCommandAdapter, ModuleSourceRoots from gyxx_flow.catalog import WorkflowEntry from gyxx_flow.core.context import RunContext from gyxx_flow.script_catalog import ScriptCatalog, ScriptEntry def _catalog(tmp_path: Path) -> ScriptCatalog: root = tmp_path / "runtime" root.mkdir() first = root / "first.py" second = root / "nested" / "second.py" second.parent.mkdir() first.write_text("print('first')\n", encoding="utf-8") second.write_text("print('second')\n", encoding="utf-8") return ScriptCatalog( ( ScriptEntry("demo:first.py", "demo", "first.py", "python", first), ScriptEntry( "demo:nested/second.py", "demo", "nested/second.py", "python", second, ), ) ) def _write_config(path: Path) -> None: path.write_text( json.dumps( { "schema_version": 1, "cdp_host": "127.0.0.1", "scripts": { "demo:first.py": 22001, "demo:nested/second.py": 22002, }, "services": { "feishu": "legacy", "postgres": "cloud", "hermes": "local", "hermes_url": "http://127.0.0.1:8642/v1/chat/completions", }, } ), encoding="utf-8", ) def test_project_registry_covers_every_script_with_unique_stable_port() -> None: project_root = Path(__file__).resolve().parents[1] scripts = ScriptCatalog.discover_default() first = RuntimeIntegrationCatalog.load( project_root / "config" / "runtime-bindings.json", scripts=scripts, data_root=project_root / "var", ) second = RuntimeIntegrationCatalog.load( project_root / "config" / "runtime-bindings.json", scripts=scripts, data_root=project_root / "var", ) assert len(scripts.script_ids) == 131 bindings = [first.binding_for(script_id) for script_id in scripts.script_ids] assert len({binding.cdp_port for binding in bindings}) == 131 assert all(22000 <= binding.cdp_port <= 22999 for binding in bindings) assert all(binding.cdp_url.startswith("http://127.0.0.1:") for binding in bindings) assert bindings == [second.binding_for(item.script_id) for item in bindings] def test_bindings_isolate_browser_state_and_relocate_with_data_root(tmp_path: Path) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_config(config) data_root = tmp_path / "portable-data" catalog = RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=data_root) first = catalog.binding_for("demo:first.py") second = catalog.binding_for("demo:nested/second.py") assert first.profile_dir != second.profile_dir assert first.cookie_file != second.cookie_file assert first.storage_state_file != second.storage_state_file for path in ( first.profile_dir, first.cookie_file, first.storage_state_file, second.profile_dir, ): assert path.is_relative_to(data_root.resolve()) assert not first.profile_dir.exists() def test_parent_and_nested_child_receive_different_runtime_environment(tmp_path: Path) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_config(config) catalog = RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=tmp_path / "data") parent = catalog.environment_for("demo:first.py", {"KEEP": "yes"}) child = catalog.environment_for("demo:nested/second.py", parent) assert child["KEEP"] == "yes" assert parent["GYXX_SCRIPT_ID"] == "demo:first.py" assert child["GYXX_SCRIPT_ID"] == "demo:nested/second.py" assert parent["GYXX_BROWSER_CDP_PORT"] != child["GYXX_BROWSER_CDP_PORT"] assert parent["GYXX_BROWSER_PROFILE_DIR"] != child["GYXX_BROWSER_PROFILE_DIR"] assert parent["GYXX_BROWSER_COOKIE_FILE"] != child["GYXX_BROWSER_COOKIE_FILE"] def test_module_command_adapter_injects_entry_script_binding(tmp_path: Path) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_config(config) catalog = RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=tmp_path / "data") runtime_root = tmp_path / "runtime" adapter = ModuleCommandAdapter( ModuleSourceRoots({"demo": runtime_root}), base_env={"KEEP": "yes"}, project_root=tmp_path, data_root=tmp_path / "data", integration_catalog=catalog, ) entry = WorkflowEntry("demo.workflow", "demo", "manual", "first.py") context = RunContext.create("demo.workflow", "2026-07-27") command = adapter.build(entry, context=context) assert command.env["KEEP"] == "yes" assert command.env["GYXX_SCRIPT_ID"] == "demo:first.py" assert command.env["GYXX_BROWSER_CDP_PORT"] == "22001" def test_cookie_store_round_trip_and_storage_state_are_atomic(tmp_path: Path) -> None: store = BrowserCookieStore( cookie_file=tmp_path / "cookies" / "cookies.json", storage_state_file=tmp_path / "cookies" / "storage_state.json", ) cookies = [{"name": "session", "value": "secret-sentinel", "domain": ".example.test", "path": "/"}] state = {"cookies": cookies, "origins": []} assert store.load_cookies() == [] assert store.load_storage_state() is None store.save_cookies(cookies) store.save_storage_state(state) assert store.load_cookies() == cookies assert store.load_storage_state() == state assert not list((tmp_path / "cookies").glob("*.tmp")) assert "secret-sentinel" not in repr(store) def test_service_policy_preserves_original_feishu_and_cloud_db_and_local_hermes() -> None: policy = RuntimeServicePolicy( hermes_url="http://127.0.0.1:8642/v1/chat/completions" ) original = { "LARK_PROFILE": "original-profile", "PG_HOST": "cloud-db.example.test", "PG_PASSWORD": "placeholder", "DB_HOST": "", "CUSTOM": "keep", } result = policy.apply(original) assert result["LARK_PROFILE"] == "original-profile" assert result["PG_HOST"] == "cloud-db.example.test" assert result["PG_PASSWORD"] == "placeholder" assert result["DB_HOST"] == "cloud-db.example.test" assert result["AUTOFLOW_PG_HOST"] == "cloud-db.example.test" assert result["GYXX_FEISHU_MODE"] == "legacy" assert result["GYXX_POSTGRES_MODE"] == "cloud" assert result["GYXX_HERMES_MODE"] == "local" assert result["HERMES_ANALYZER_URL"].startswith("http://127.0.0.1:") assert "secret-sentinel" not in repr(policy) @pytest.mark.parametrize( ("environment", "message"), [ ({"PG_HOST": "127.0.0.1"}, "cloud"), ({"DB_HOST": "localhost"}, "cloud"), ({"AUTOFLOW_PG_HOST": "::1"}, "cloud"), ({"HERMES_ANALYZER_URL": "https://remote.example.test/v1"}, "local"), ({"ANALYZER_API_SERVER_URL": "http://10.0.0.8:8642/v1"}, "local"), ], ) def test_service_policy_rejects_local_database_or_remote_hermes( environment: dict[str, str], message: str ) -> None: with pytest.raises(RuntimeIntegrationError, match=message): RuntimeServicePolicy().apply(environment) def test_registry_rejects_missing_duplicate_or_unknown_script_allocations(tmp_path: Path) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_config(config) payload = json.loads(config.read_text(encoding="utf-8")) payload["scripts"]["demo:nested/second.py"] = 22001 config.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(RuntimeIntegrationError, match="unique"): RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=tmp_path / "data") payload["scripts"].pop("demo:nested/second.py") payload["scripts"]["demo:unknown.py"] = 22003 config.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(RuntimeIntegrationError, match="coverage"): RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=tmp_path / "data") def test_nested_script_browser_arguments_are_rebound(monkeypatch, tmp_path: Path) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_config(config) binding = RuntimeIntegrationCatalog.load( config, scripts=scripts, data_root=tmp_path / "data" ).binding_for("demo:nested/second.py") monkeypatch.setattr( "sys.argv", [ "second.py", "--user-data-dir", "parent-profile", "--cdp-url=http://127.0.0.1:22001", "--cdp-port", "22001", ], ) _rewrite_browser_arguments(binding) assert sys.argv[2] == str(binding.profile_dir) assert sys.argv[3] == f"--cdp-url={binding.cdp_url}" assert sys.argv[5] == str(binding.cdp_port)