346 lines
13 KiB
Python
346 lines
13 KiB
Python
#!/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()
|