from __future__ import annotations import json import os import subprocess import sys from pathlib import Path import pytest from gyxx_flow.adapters import ( BrowserCookieStore, RuntimeIntegrationCatalog, RuntimeIntegrationError, RuntimeServicePolicy, binding_from_environment, environment_for_child_script, resolve_hermes_profile_api_key, ) from gyxx_flow.adapters.bootstrap import ( _rewrite_browser_arguments, bootstrap_current_process, ) 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", "demo", "first.py", "python", first), ScriptEntry( "demo.second", "demo", "nested/second.py", "python", second, ), ) ) def _write_config(path: Path) -> None: path.write_text( json.dumps( { "schema_version": 2, "cdp_host": "127.0.0.1", "scripts": { "demo.first": { "script_id": "demo:first.py", "state_key": "demo:first.py", "aliases": ["demo:first.py"], "cdp_port": 22001, "login_mode": "C", "required_cookie_domains": ["example.test"], "required_cookie_names": ["session"], "credential_env_names": ["DEMO_ACCOUNT", "DEMO_PASSWORD"], }, "demo.second": { "script_id": "demo:nested/second.py", "state_key": "demo:nested/second.py", "aliases": ["demo:nested/second.py"], "cdp_port": 22002, }, }, "services": { "feishu": "legacy", "postgres": "cloud", "hermes": "local", "hermes_url": "http://127.0.0.1:8642/v1/chat/completions", "hermes_collector_url": "http://127.0.0.1:8643/v1/chat/completions", "hermes_analyzer_gateway_url": "http://127.0.0.1:8642/v1", "hermes_collector_gateway_url": "http://127.0.0.1:8643/v1", }, } ), 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.command_ids) == 49 assert len(first.command_ids) == 152 bindings = [first.binding_for(command_id) for command_id in first.command_ids] assert len({binding.cdp_port for binding in bindings}) == 152 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] tmall_ads = first.binding_for("product.import.tmall_ads") assert tmall_ads is first.binding_for( "product_commerce:import_tmall_product_ads.py" ) assert tmall_ads.cdp_port == 22140 appeal = first.binding_for("shop.douyin_price_appeal") assert appeal is first.binding_for( "shop_intelligence:collectors/dy_store_competitor_store_scraping.py" ) assert appeal.cdp_port == 22104 video_upload = first.binding_for("product.video_upload.run") assert video_upload is first.binding_for( "product_commerce:upload_video_to_guanghe.py" ) assert video_upload.cdp_port == 22098 assert video_upload.login_mode == "C" assert video_upload.required_cookie_domains == ("taobao.com",) assert video_upload.credential_env_names == ( "GUANGHE_USERNAME", "GUANGHE_PASSWORD", "GUANGHE_LUGGAGE_USERNAME", "GUANGHE_LUGGAGE_PASSWORD", ) video_environment = first.environment_for("product.video_upload.run", {}) assert video_environment["GUANGHE_USER_DATA_DIR"] == str( video_upload.profile_dir ) assert video_environment["GUANGHE_LUGGAGE_USER_DATA_DIR"] != str( video_upload.profile_dir ) assert video_environment["GUANGHE_LUGGAGE_USER_DATA_DIR"].endswith( "-luggage" ) jd_video_upload = first.binding_for("product.jd_video_upload.run") assert jd_video_upload is first.binding_for( "product_commerce:upload_video_to_jd.py" ) assert jd_video_upload.cdp_port == 22137 assert jd_video_upload.login_mode == "C" assert jd_video_upload.required_cookie_domains == ("jd.com",) assert jd_video_upload.credential_env_names == ("JD_SHOP", "JD_PASSWORD") assert jd_video_upload.profile_dir != video_upload.profile_dir erp_all_shop = first.binding_for("product.erp_all_shop_daily") assert erp_all_shop.cdp_port == 22138 assert erp_all_shop.required_cookie_domains == ("erp321.com",) assert jd_video_upload.cookie_file != video_upload.cookie_file assert jd_video_upload.storage_state_file != video_upload.storage_state_file baibu_old = first.binding_for("product.tmall_baibu_apply.old_chain_bag") assert baibu_old.account == "tmall-shop" assert baibu_old.cdp_port == 22148 assert baibu_old.required_cookie_domains == ("taobao.com", "alimama.com") baibu_new = first.binding_for("product.tmall_baibu_apply.new_chain_bag") assert baibu_new.account == "tmall-bag" assert baibu_new.cdp_port == 22152 assert baibu_new.login_mode == "C" assert baibu_new.credential_env_names == ( "TMALL_BAG_ACCOUNT", "TMALL_BAG_PASSWORD", ) assert baibu_new.profile_dir != baibu_old.profile_dir jd_ad_costs = first.binding_for("product.jd_ad_costs.collect") assert jd_ad_costs is first.binding_for( "product_commerce:collect_jd_ad_costs.py" ) assert jd_ad_costs.cdp_port == 22144 assert jd_ad_costs.account == "jd-shop" assert jd_ad_costs.required_cookie_domains == ("jd.com",) collaborator_bilibili = first.binding_for( "content_marketing:bilibili_scraper.py" ) self_bilibili = first.binding_for( "content_marketing:self_bilibili_scraper.py" ) for binding in (collaborator_bilibili, self_bilibili): assert binding.required_cookie_domains == () assert binding.required_cookie_names == () comment_bilibili = first.binding_for("content.comments.collect_bilibili") assert comment_bilibili.required_cookie_domains == ("bilibili.com",) assert comment_bilibili.required_cookie_names == ("SESSDATA",) 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 _write_account_config(path: Path) -> None: path.write_text( json.dumps( { "schema_version": 3, "cdp_host": "127.0.0.1", "accounts": { "demo-shop": { "cdp_port": 22200, "login_mode": "D", "required_cookie_domains": ["example.test"], "required_cookie_names": ["session", "sid_tt"], } }, "scripts": { "demo.first": { "script_id": "demo:first.py", "state_key": "demo:first.py", "aliases": ["demo:first.py"], "cdp_port": 22001, "login_mode": "A", "required_cookie_domains": ["example.test"], "account": "demo-shop", }, "demo.second": { "script_id": "demo:nested/second.py", "state_key": "demo:nested/second.py", "aliases": ["demo:nested/second.py"], "cdp_port": 22002, }, }, "services": { "feishu": "legacy", "postgres": "cloud", "hermes": "local", "hermes_url": "http://127.0.0.1:8642/v1/chat/completions", "hermes_collector_url": "http://127.0.0.1:8643/v1/chat/completions", "hermes_analyzer_gateway_url": "http://127.0.0.1:8642/v1", "hermes_collector_gateway_url": "http://127.0.0.1:8643/v1", }, } ), encoding="utf-8", ) def test_accounts_parse_and_sync_vault_into_member_bindings(tmp_path: Path) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_account_config(config) data_root = tmp_path / "data" catalog = RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=data_root) account = catalog.account_for("demo-shop") assert account.cdp_port == 22200 assert account.login_mode == "D" assert account.required_cookie_names == ("session", "sid_tt") assert account.cookie_file == ( data_root / "state" / "accounts" / "demo-shop" / "cookies.json" ) members = catalog.bindings_for_account("demo-shop") assert [member.command_id for member in members] == ["demo.first"] assert catalog.binding_for("demo.second").account == "" environment = catalog.environment_for("demo:first.py", {}) assert environment["GYXX_ACCOUNT_ID"] == "demo-shop" assert environment["GYXX_ACCOUNT_PROFILE_DIR"] == str(account.profile_dir) assert environment["GYXX_ACCOUNT_COOKIE_FILE"] == str(account.cookie_file) dynamic = catalog.environment_for_account("demo-shop", {"KEEP": "yes"}) assert dynamic["KEEP"] == "yes" assert dynamic["GYXX_ACCOUNT_STORAGE_STATE_FILE"] == str( account.storage_state_file ) with pytest.raises(RuntimeIntegrationError, match="unknown runtime account"): catalog.account_for("nope") vault_cookies = [ { "name": "session", "value": "VAULT_SESSION", "domain": "example.test", "path": "/", }, {"name": "sid_tt", "value": "VAULT_SID", "domain": "example.test", "path": "/"}, ] BrowserCookieStore(account.cookie_file, account.storage_state_file).save_cookies( vault_cookies ) BrowserCookieStore( account.cookie_file, account.storage_state_file ).save_storage_state( { "cookies": vault_cookies, "origins": [ { "origin": "https://example.test", "localStorage": [{"name": "k", "value": "v"}], } ], } ) member = catalog.binding_for("demo.first") member_store = BrowserCookieStore(member.cookie_file, member.storage_state_file) member_store.save_cookies( [ { "name": "session", "value": "OLD_SESSION", "domain": "example.test", "path": "/", }, {"name": "local_only", "value": "keep", "domain": "other.test", "path": "/"}, ] ) member_store.save_storage_state( { "cookies": [], "origins": [ { "origin": "https://legacy.test", "localStorage": [{"name": "a", "value": "b"}], } ], } ) synced = catalog.sync_binding_state(member) assert synced["synced"] is True merged = member_store.load_cookies() by_key = { (item["name"], item["domain"]): item["value"] for item in merged } assert by_key[("session", "example.test")] == "VAULT_SESSION" assert by_key[("sid_tt", "example.test")] == "VAULT_SID" assert by_key[("local_only", "other.test")] == "keep" state = member_store.load_storage_state() origins = {item["origin"] for item in state["origins"]} assert origins == {"https://example.test", "https://legacy.test"} # A second sync with an unchanged vault is a no-op. assert catalog.sync_binding_state(member)["synced"] is False # environment_for pushes the vault before returning the environment. catalog.environment_for("demo.first", {}) assert catalog.binding_for("demo.first").cookie_file.read_text( encoding="utf-8" ).count("VAULT_SESSION") == 1 def test_account_unknown_reference_and_port_collisions_are_rejected( tmp_path: Path, ) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_account_config(config) payload = json.loads(config.read_text(encoding="utf-8")) payload["scripts"]["demo.first"]["account"] = "missing-account" config.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(RuntimeIntegrationError, match="unknown runtime account"): RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=tmp_path / "d") _write_account_config(config) payload = json.loads(config.read_text(encoding="utf-8")) payload["accounts"]["demo-shop"]["cdp_port"] = 22001 # collides with demo.first config.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(RuntimeIntegrationError, match="unique integers"): RuntimeIntegrationCatalog.load(config, scripts=scripts, data_root=tmp_path / "d") def test_project_account_pilot_covers_douyin_shop_members() -> None: project_root = Path(__file__).resolve().parents[1] catalog = RuntimeIntegrationCatalog.load( project_root / "config" / "runtime-bindings.json", scripts=ScriptCatalog.discover_default(), data_root=project_root / "var", ) account = catalog.account_for("douyin-shop") assert account.cdp_port == 22200 members = catalog.bindings_for_account("douyin-shop") assert {member.command_id for member in members} == { "shop.douyin_price_appeal", "product_commerce:dy_product_scraping.py", "product_commerce:dy_audience_profile_collect.py", "product_commerce:collect_dy_market_rank.py", "product.douyin_qianchuan_ads.collect", "product_commerce:vendors/dy-data-flow/dy_store_competitor_store_scraping.py", "content_marketing:xingtu_scraper_v2.py", } for member in members: assert member.required_cookie_domains == ("jinritemai.com",) def test_project_accounts_cover_all_configured_platform_login_identities() -> None: project_root = Path(__file__).resolve().parents[1] catalog = RuntimeIntegrationCatalog.load( project_root / "config" / "runtime-bindings.json", scripts=ScriptCatalog.discover_default(), data_root=project_root / "var", ) expected_delays = { "douyin-shop": 0, "jd-shop": 60, "jd-self-operated": 120, "jd-market-rank": 540, "erp-jushuitan": 660, "tmall-shop": 180, "tmall-ozko": 600, "tmall-bag": 300, "xiaohongshu-content": 240, "douyin-content": 300, "xingtu": 360, "chanmama": 420, "bilibili-content": 480, } assert {account.account_id for account in catalog.accounts} == set(expected_delays) assert { account.account_id: account.keepalive_initial_delay_seconds for account in catalog.accounts } == expected_delays assert all(account.keepalive_enabled for account in catalog.accounts) assert all(account.keepalive_real_chrome for account in catalog.accounts) assert all( catalog.bindings_for_account(account.account_id) for account in catalog.accounts if account.account_id != "tmall-ozko" ) assert catalog.bindings_for_account("tmall-ozko") == () assert catalog.account_for("jd-shop").credential_env_names == ( "JD_SHOP", "JD_PASSWORD", ) assert catalog.account_for("jd-self-operated").credential_env_names == ( "JD_SELF_OPERATED_ACCOUNT", "JD_SELF_OPERATED_PASSWORD", ) assert catalog.account_for("tmall-shop").credential_env_names == ( "TMALL_SHOP_ACCOUNT", "TMALL_SHOP_PASSWORD", "SYCM_ACCOUNT", "SYCM_PASSWORD", "WANXIANG_ACCOUNT", "WANXIANG_PASSWORD", ) assert catalog.account_for("tmall-shop").login_identity_markers == ( "光影行星旗舰店", ) assert catalog.account_for("tmall-bag").login_identity_markers == ( "光影行星鑫华达专卖店", ) assert catalog.account_for("erp-jushuitan").credential_env_names == ( "ERP_USERNAME", "ERP_PASSWORD", "GYXX_SUPPLY_ERP_PASSWORD", ) assert catalog.account_for("erp-jushuitan").login_command_id == ( "product_commerce:erp_login_product_analysis.py" ) jd_market = catalog.binding_for("product_commerce:collect_jd_market_rank.py") assert jd_market.account == "jd-market-rank" assert jd_market.required_cookie_names == ("pin", "thor") assert jd_market.credential_env_names == ( "JD_MARKET_SHOP", "JD_MARKET_PASSWORD", ) erp_daily = catalog.binding_for( "product_commerce:collect_erp_yesterday_metrics.py" ) assert erp_daily.account == "erp-jushuitan" assert catalog.binding_for("product.erp_all_shop_daily").account == "erp-jushuitan" assert catalog.binding_for("product.sales_sheet.sync").account == "erp-jushuitan" assert catalog.binding_for("supply.replenishment.arrival_sync").account == "erp-jushuitan" def test_schema_v1_keeps_legacy_script_ids_and_existing_state_paths(tmp_path: Path) -> None: scripts = _catalog(tmp_path) stable_config = tmp_path / "stable.json" legacy_config = tmp_path / "legacy.json" _write_config(stable_config) payload = json.loads(stable_config.read_text(encoding="utf-8")) payload["schema_version"] = 1 payload["scripts"] = { item["script_id"]: item["cdp_port"] for item in payload["scripts"].values() } legacy_config.write_text(json.dumps(payload), encoding="utf-8") stable = RuntimeIntegrationCatalog.load( stable_config, scripts=scripts, data_root=tmp_path / "data" ) legacy = RuntimeIntegrationCatalog.load( legacy_config, scripts=scripts, data_root=tmp_path / "data" ) assert stable.binding_for("demo.first").profile_dir == legacy.binding_for( "demo:first.py" ).profile_dir assert stable.binding_for("demo.first").cookie_file == legacy.binding_for( "demo:first.py" ).cookie_file def test_state_key_survives_a_physical_entry_move_and_old_alias_still_resolves( tmp_path: Path, ) -> None: scripts = _catalog(tmp_path) config = tmp_path / "bindings.json" _write_config(config) before = RuntimeIntegrationCatalog.load( config, scripts=scripts, data_root=tmp_path / "data" ).binding_for("demo.first") payload = json.loads(config.read_text(encoding="utf-8")) payload["scripts"]["demo.first"]["script_id"] = "demo:moved/first.py" config.write_text(json.dumps(payload), encoding="utf-8") moved_path = tmp_path / "runtime" / "moved" / "first.py" moved_path.parent.mkdir() moved_path.write_text("print('moved')\n", encoding="utf-8") moved_scripts = ScriptCatalog( (ScriptEntry("demo.first", "demo", "moved/first.py", "python", moved_path),) ) catalog = RuntimeIntegrationCatalog.load( config, scripts=moved_scripts, data_root=tmp_path / "data" ) after = catalog.binding_for("demo.first") assert catalog.binding_for("demo:first.py") is after assert catalog.binding_for("demo:moved/first.py") is after assert after.profile_dir == before.profile_dir assert after.cookie_file == before.cookie_file assert after.storage_state_file == before.storage_state_file 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_COMMAND_ID"] == "demo.first" assert child["GYXX_COMMAND_ID"] == "demo.second" assert parent["GYXX_BROWSER_STATE_KEY"] == "demo:first.py" assert parent["GYXX_BROWSER_LOGIN_MODE"] == "C" assert json.loads(parent["GYXX_BROWSER_REQUIRED_COOKIE_DOMAINS"]) == [ "example.test" ] assert json.loads(parent["GYXX_BROWSER_REQUIRED_COOKIE_NAMES"]) == ["session"] assert json.loads(parent["GYXX_BROWSER_CREDENTIAL_ENV_NAMES"]) == [ "DEMO_ACCOUNT", "DEMO_PASSWORD", ] assert parent["DY_COOKIES_FILE"] == parent["GYXX_BROWSER_COOKIE_FILE"] 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"] reconstructed = binding_from_environment(parent) assert reconstructed.login_mode == "C" assert reconstructed.required_cookie_domains == ("example.test",) assert reconstructed.required_cookie_names == ("session",) assert reconstructed.credential_env_names == ("DEMO_ACCOUNT", "DEMO_PASSWORD") def test_private_nested_child_keeps_legacy_alias_binding(tmp_path: Path) -> None: project_root = Path(__file__).resolve().parents[1] module_root = ( project_root / "src" / "gyxx_flow" / "modules" / "product_commerce" / "runtime" ) environment = environment_for_child_script( module_root / "taobao_sycm_products.py", { "GYXX_MODULE_ID": "product_commerce", "GYXX_MODULE_ROOT": str(module_root), "GYXX_PROJECT_ROOT": str(project_root), "GYXX_DATA_ROOT": str(tmp_path / "data"), }, ) assert environment["GYXX_SCRIPT_ID"] == "product_commerce:taobao_sycm_products.py" assert environment["GYXX_COMMAND_ID"] == "product_commerce:taobao_sycm_products.py" assert environment["GYXX_BROWSER_STATE_KEY"] == ( "product_commerce:taobao_sycm_products.py" ) assert environment["GYXX_BROWSER_CDP_PORT"] == "22096" def test_jd_self_operated_collectors_consume_their_own_browser_binding( tmp_path: Path, ) -> None: project_root = Path(__file__).resolve().parents[1] scripts = ScriptCatalog.discover_default() catalog = RuntimeIntegrationCatalog.load( project_root / "config" / "runtime-bindings.json", scripts=scripts, data_root=tmp_path / "portable-data", ) module_root = ( project_root / "src" / "gyxx_flow" / "modules" / "shop_intelligence" ) cases = ( ( "shop_intelligence:collectors/jd_self_operated_brand_daily.py", "gyxx_flow.modules.shop_intelligence.collectors.jd_self_operated_brand_daily", ), ( "shop_intelligence:collectors/jd_self_operated_product_daily.py", "gyxx_flow.modules.shop_intelligence.collectors.jd_self_operated_product_daily", ), ) observed_ports: list[int] = [] for script_id, module_name in cases: binding = catalog.binding_for(script_id) environment = catalog.environment_for(script_id, os.environ) environment["PYTHONPATH"] = os.pathsep.join( ( str(project_root / "src"), str(module_root), environment.get("PYTHONPATH", ""), ) ) command = ( "import json; " f"import {module_name} as module; " "print(json.dumps({" "'port': module.CDP_PORT, " "'profile': str(module.BROWSER_PROFILE_DIR), " "'storage': str(module.STORAGE_STATE_PATH)" "}))" ) result = subprocess.run( [sys.executable, "-c", command], cwd=module_root, env=environment, capture_output=True, text=True, encoding="utf-8", timeout=30, check=False, ) assert result.returncode == 0, result.stderr payload = json.loads(result.stdout.strip().splitlines()[-1]) assert payload == { "port": binding.cdp_port, "profile": str(binding.profile_dir), "storage": str(binding.storage_state_file), } observed_ports.append(payload["port"]) assert observed_ports == [22132, 22133] 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_COMMAND_ID"] == "demo.first" assert command.env["GYXX_BROWSER_STATE_KEY"] == "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_cookie_store_reads_legacy_cookie_envelope(tmp_path: Path) -> None: store = BrowserCookieStore( cookie_file=tmp_path / "cookies.json", storage_state_file=tmp_path / "storage_state.json", ) cookies = [{"name": "session", "value": "legacy", "domain": ".example.test"}] store.cookie_file.parent.mkdir(parents=True, exist_ok=True) store.cookie_file.write_text( json.dumps({"cookies": cookies, "timestamp": "legacy"}), encoding="utf-8", ) assert store.load_cookies() == cookies def test_service_policy_preserves_feishu_and_uses_cloud_pg_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_PORT": "5432", "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["DB_PORT"] == "5432" assert result["AUTOFLOW_PG_PORT"] == "5432" 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 result["COLLECTOR_API_SERVER_URL"].startswith("http://127.0.0.1:") assert result["ANALYZER_HERMES_GATEWAY_URL"].endswith(":8642/v1") assert result["COLLECTOR_HERMES_GATEWAY_URL"].endswith(":8643/v1") assert "secret-sentinel" not in repr(policy) @pytest.mark.parametrize( ("environment", "message"), [ ({"PG_HOST": "127.0.0.1"}, "remote"), ({"DB_HOST": "localhost"}, "remote"), ({"AUTOFLOW_PG_HOST": "::1"}, "remote"), ({"DATABASE_URL": "postgresql://user@127.0.0.1/gyxx"}, "remote"), ({"HERMES_ANALYZER_URL": "https://remote.example.test/v1"}, "local"), ({"ANALYZER_API_SERVER_URL": "http://10.0.0.8:8642/v1"}, "local"), ({"COLLECTOR_HERMES_GATEWAY_URL": "https://remote.example.test/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_service_policy_supplies_portable_local_pg_defaults_without_password() -> None: result = RuntimeServicePolicy( postgres_mode="local", postgres_host="127.0.0.1", postgres_database="gyxx_super_data", postgres_user="gyxx_flow", ).apply({}) assert result["PG_HOST"] == "127.0.0.1" assert result["DB_HOST"] == "127.0.0.1" assert result["AUTOFLOW_PG_HOST"] == "127.0.0.1" assert result["PG_PORT"] == "5432" assert result["DB_PORT"] == "5432" assert result["AUTOFLOW_PG_PORT"] == "5432" assert result["PG_DB"] == "gyxx_super_data" assert result["DB_NAME"] == "gyxx_super_data" assert result["PG_USER"] == "gyxx_flow" assert "PG_PASSWORD" not in result def test_service_policy_maps_canonical_secrets_without_exposing_defaults() -> None: result = RuntimeServicePolicy().apply( { "GYXX_POSTGRES_PASSWORD": "placeholder", "GYXX_HERMES_API_KEY": "placeholder", } ) assert result["PG_PASSWORD"] == "placeholder" assert result["DB_PASSWORD"] == "placeholder" assert result["AUTOFLOW_PG_PASSWORD"] == "placeholder" assert result["HERMES_API_KEY"] == "placeholder" assert result["HERMES_ANALYZER_TOKEN"] == "placeholder" assert result["GYXX_SUPPLY_HERMES_TOKEN"] == "placeholder" def test_shared_hermes_resolver_reads_runtime_profile_without_copying_secret( tmp_path: Path, ) -> None: profile = tmp_path / "profiles" / "data-analyzer" profile.mkdir(parents=True) (profile / ".env").write_text( "API_SERVER_PORT=8642\nAPI_SERVER_KEY=local-only-key\n", encoding="utf-8", ) assert resolve_hermes_profile_api_key( "data-analyzer", {"HERMES_HOME": str(tmp_path)}, ) == "local-only-key" def test_shared_hermes_resolver_prefers_explicit_runtime_secret(tmp_path: Path) -> None: assert resolve_hermes_profile_api_key( "data-analyzer", { "HERMES_HOME": str(tmp_path), "HERMES_ANALYZER_TOKEN": "explicit-key", }, preferred_environment_names=("HERMES_ANALYZER_TOKEN",), ) == "explicit-key" def test_service_policy_replaces_present_but_blank_local_pg_aliases() -> None: result = RuntimeServicePolicy( postgres_mode="local", postgres_host="127.0.0.1", postgres_database="gyxx_super_data", postgres_user="gyxx_flow", ).apply( {"PG_HOST": "", "DB_PORT": " ", "AUTOFLOW_PG_DB": ""} ) assert result["PG_HOST"] == "127.0.0.1" assert result["DB_PORT"] == "5432" assert result["AUTOFLOW_PG_DB"] == "gyxx_super_data" def test_service_policy_maps_cloud_dsn_to_all_database_aliases() -> None: result = RuntimeServicePolicy().apply( {"GYXX_POSTGRES_DSN": "postgresql://cloud_user:placeholder@db.example.test/data_hub"} ) assert result["GYXX_POSTGRES_MODE"] == "cloud" assert result["PG_HOST"] == "db.example.test" assert result["DB_HOST"] == "db.example.test" assert result["AUTOFLOW_PG_HOST"] == "db.example.test" assert result["PG_DB"] == "data_hub" assert result["AUTOFLOW_PG_USER"] == "cloud_user" 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.second"]["cdp_port"] = 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.second") payload["scripts"]["demo.unknown"] = { "script_id": "demo:unknown.py", "state_key": "demo:unknown.py", "aliases": ["demo:unknown.py"], "cdp_port": 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) def test_parent_routed_child_preserves_dynamic_runtime_environment( monkeypatch, 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" ) entry = scripts.get("demo.second") monkeypatch.setattr( "gyxx_flow.adapters.bootstrap.RuntimeIntegrationCatalog.load_default", lambda **_kwargs: catalog, ) monkeypatch.setenv("GYXX_MODULE_ID", "demo") monkeypatch.setenv("GYXX_MODULE_ROOT", str(tmp_path / "runtime")) monkeypatch.setenv("GYXX_PROJECT_ROOT", str(tmp_path)) monkeypatch.setenv("GYXX_DATA_ROOT", str(tmp_path / "data")) monkeypatch.setenv("GYXX_PRESERVE_RUNTIME_ENV", "1") monkeypatch.setenv("GYXX_BROWSER_CDP_PORT", "22143") monkeypatch.setenv("GYXX_BROWSER_CDP_URL", "http://127.0.0.1:22143") monkeypatch.setattr( sys, "argv", [ str(entry.path), "--cdp-url", "http://127.0.0.1:22143", ], ) binding = bootstrap_current_process() assert binding is catalog.binding_for("demo:nested/second.py") assert os.environ["GYXX_BROWSER_CDP_PORT"] == "22143" assert sys.argv[2] == "http://127.0.0.1:22143"