#!/usr/bin/env python3 from __future__ import annotations import json import os import sys import tarfile import tempfile import unittest from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] CONFIG_SRC = PROJECT_ROOT / "packages" / "jwebot-config" / "src" sys.path.insert(0, str(CONFIG_SRC)) from jwebot_config import ( # noqa: E402 DEFAULT_PROFILE_FILES, ConfigError, ConfigLayout, activate_profile, archive_legacy_material, compile_profile, export_profile, import_profile, load_active_environment, load_private_context, migrate_legacy_context, validate_profile, write_default_profile, ) class ConfigProfileTests(unittest.TestCase): def layout(self, root: Path) -> ConfigLayout: return ConfigLayout( config_root=root / "config", state_root=root / "state", ) def test_repository_template_matches_generated_defaults(self): template = PROJECT_ROOT / "config" / "templates" / "default" for relative, expected in DEFAULT_PROFILE_FILES.items(): with self.subTest(path=relative): self.assertEqual( (template / relative).read_text(encoding="utf-8"), expected, ) def test_default_profile_compiles_and_activates_atomically(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) layout = self.layout(root) profile = write_default_profile("workstation", layout) self.assertEqual(validate_profile(profile), "workstation") generated = compile_profile(profile, layout) activate_profile("workstation", layout) self.assertTrue(layout.current_link.is_symlink()) self.assertEqual(layout.active_profile(), profile.resolve()) self.assertEqual( os.stat(generated / "runtime.env").st_mode & 0o777, 0o600, ) for directory_path in ( layout.config_root, layout.profiles_root, layout.state_root, layout.state_root / "profiles", layout.profile_state("workstation"), ): self.assertEqual( os.stat(directory_path).st_mode & 0o777, 0o700, ) environment = load_active_environment(layout) self.assertEqual(environment["JWEBOT_CONFIG_PROFILE"], "workstation") self.assertEqual(environment["JWEBOT_EDGE_ENABLED"], "1") self.assertEqual(environment["JWEBOT_EDGE_STEAM_API_KEY"], "") self.assertEqual( load_private_context(layout)["bot"]["mention"], "JWeBot", ) def test_private_context_and_component_values_are_compiled_separately(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) layout = self.layout(root) profile = write_default_profile("custom", layout) (profile / "context" / "people.json").write_text( json.dumps( { "people": [ { "canonical_name": "Alice", "aliases": ["alice-ocr"], "prefixes": [], "identity_fact": "", "display_id": "alice", "steam_id64": "10000000000000001", "steam_aliases": ["alice-steam"], } ], "clock_scan_targets": [], } ), encoding="utf-8", ) (profile / "secrets" / "runtime.env").write_text( "JWEBOT_EDGE_DEEPSEEK_API_KEY=test-secret\n" "JWEBOT_EDGE_STEAM_API_KEY=\n", encoding="utf-8", ) generated = compile_profile(profile, layout) environment = (generated / "runtime.env").read_text(encoding="utf-8") context = json.loads( (generated / "private-context.json").read_text(encoding="utf-8") ) self.assertIn("JWEBOT_EDGE_DEEPSEEK_API_KEY=test-secret", environment) self.assertNotIn("alice-ocr", environment) self.assertEqual(context["people"][0]["display_id"], "alice") self.assertEqual( context["people"][0]["steam_id64"], "10000000000000001", ) self.assertEqual( context["people"][0]["steam_aliases"], ["alice-steam"], ) self.assertNotIn("test-secret", json.dumps(context)) def test_redacted_export_removes_secrets_and_people(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) layout = self.layout(root) profile = write_default_profile("shareable", layout) (profile / "secrets" / "runtime.env").write_text( "JWEBOT_EDGE_DEEPSEEK_API_KEY=test-secret\n" "JWEBOT_EDGE_STEAM_API_KEY=another-secret\n", encoding="utf-8", ) (profile / "context" / "people.json").write_text( '{"people":[{"canonical_name":"Alice","aliases":["a"],' '"prefixes":[],"identity_fact":"","display_id":"a"}],' '"clock_scan_targets":[]}\n', encoding="utf-8", ) output = root / "shareable.tar.gz" export_profile("shareable", output, layout) with tarfile.open(output, "r:gz") as archive: secret_text = archive.extractfile( "shareable/secrets/runtime.env" ).read().decode() context_text = archive.extractfile( "shareable/context/people.json" ).read().decode() self.assertNotIn("test-secret", secret_text) self.assertNotIn("another-secret", secret_text) self.assertEqual(json.loads(context_text)["people"], []) def test_private_context_rejects_duplicate_steam_ids(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) layout = self.layout(root) profile = write_default_profile("duplicate-steam", layout) people = [] for name in ("Alice", "Bob"): people.append( { "canonical_name": name, "aliases": [], "prefixes": [], "identity_fact": "", "display_id": name.casefold(), "steam_id64": "10000000000000001", "steam_aliases": [], } ) (profile / "context" / "people.json").write_text( json.dumps( {"people": people, "clock_scan_targets": []} ), encoding="utf-8", ) with self.assertRaisesRegex(ConfigError, "must be unique"): validate_profile(profile) def test_import_rejects_archive_path_traversal(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) layout = self.layout(root) archive_path = root / "unsafe.tar" payload = root / "payload" payload.write_text("value", encoding="utf-8") with tarfile.open(archive_path, "w") as archive: archive.add(payload, arcname="../outside") with self.assertRaisesRegex(ConfigError, "unsafe path"): import_profile(archive_path, layout) def test_active_profile_rejects_symlink_outside_profile_root(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) layout = self.layout(root) layout.config_root.mkdir(parents=True) outside = root / "outside" outside.mkdir() layout.current_link.symlink_to(outside) with self.assertRaisesRegex(ConfigError, "outside"): layout.active_profile() def test_legacy_debug_material_is_archived_outside_the_repository(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) repository = root / "repository" repository.mkdir() (repository / ".dbg").mkdir() (repository / ".dbg" / "session.env").write_text( "DEBUG_VALUE=local\n", encoding="utf-8", ) (repository / "debug-incident.md").write_text( "private incident", encoding="utf-8", ) layout = self.layout(root) write_default_profile("migrated", layout) archive = archive_legacy_material( repository, "migrated", layout, ) self.assertTrue((archive / ".dbg" / "session.env").is_file()) self.assertTrue((archive / "debug-incident.md").is_file()) self.assertEqual( os.stat(archive / ".dbg" / "session.env").st_mode & 0o777, 0o600, ) def test_legacy_context_is_migrated_into_private_profile(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) repository = root / "repository" source = ( repository / "packages" / "jwebot-kernel" / "src" / "jwebot_kernel" ) source.mkdir(parents=True) (source / "edge_runtime.py").write_text( "\n".join( ( 'BOT_IDENTITY_NOTE = "ExampleBot identity"', "PERSON_ALIAS_TEMPLATES = (", " {", ' "canonical_name": "Alice",', ' "aliases": ("alice-ocr",),', ' "prefixes": (),', ' "identity_fact": "Alice is a test user",', " },", " {", ' "canonical_name": "Bob",', ' "aliases": ("bob-ocr",),', ' "prefixes": (),', ' "identity_fact": "Bob is a test user",', " },", ")", 'PERSON_DISPLAY_IDS = {"Alice": "alice", "Bob": "bob"}', 'CLOCK_SCAN_TARGETS = (("Alice", "alice"), ("Bob", "bob"))', ) ) + "\n", encoding="utf-8", ) steam_source = ( repository / "packages" / "jwebot-official-extensions" / "src" / "jwebot_official_extensions" ) steam_source.mkdir(parents=True) (steam_source / "steam.py").write_text( "\n".join( ( 'STEAM_GROUP_ROSTER = (("bob", "10000000000000002"),)', "STEAM_ALIAS_LABELS = {", ' "alice-steam": "alice",', ' "bob-steam": "bob",', "}", ) ) + "\n", encoding="utf-8", ) environment = root / "environment" environment.write_text( "JWEBOT_EDGE_STEAM_ID64=10000000000000001\n", encoding="utf-8", ) layout = self.layout(root) profile = write_default_profile("migrated", layout) self.assertTrue( migrate_legacy_context( repository, "migrated", layout, environment=environment, ) ) context = json.loads( (profile / "context" / "people.json").read_text(encoding="utf-8") ) bot = (profile / "components" / "bot.toml").read_text( encoding="utf-8" ) self.assertEqual(context["people"][0]["display_id"], "alice") self.assertEqual(context["clock_scan_targets"][0]["display_id"], "alice") self.assertEqual( [person["steam_id64"] for person in context["people"]], ["10000000000000001", "10000000000000002"], ) self.assertEqual( [person["steam_aliases"] for person in context["people"]], [["alice-steam"], ["bob-steam"]], ) self.assertIn('mention = "ExampleBot"', bot) if __name__ == "__main__": unittest.main()