mirror of
https://github.com/seanbetts/steam-hardware-watch.git
synced 2026-10-06 01:00:18 +02:00
245 lines
9.3 KiB
Python
245 lines
9.3 KiB
Python
import os
|
|
import subprocess
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class RunWatchCliTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.repo = Path(self.tmp.name)
|
|
self.scripts = self.repo / "scripts"
|
|
self.scripts.mkdir()
|
|
(self.repo / "status" / "runs").mkdir(parents=True)
|
|
self.log = self.repo / "calls.log"
|
|
|
|
os.symlink(ROOT / "scripts" / "run_watch.sh", self.scripts / "run_watch.sh")
|
|
os.symlink(ROOT / "scripts" / "run_frame_watch.sh", self.scripts / "run_frame_watch.sh")
|
|
|
|
self._write_script(
|
|
"init_run.sh",
|
|
"""
|
|
#!/bin/sh
|
|
set -eu
|
|
run_date="$1"
|
|
base_dir="$2"
|
|
run_dir="$base_dir/$run_date"
|
|
mkdir -p "$run_dir/reports" "$run_dir/api"
|
|
run_note="$PWD/status/runs/$run_date.md"
|
|
printf 'init_run %s %s\\n' "$run_date" "$base_dir" >> "$CALL_LOG"
|
|
printf 'RUN_DIR=%s\\nRUN_NOTE=%s\\n' "$run_dir" "$run_note"
|
|
""",
|
|
)
|
|
self._write_script(
|
|
"check_komodo.sh",
|
|
"""
|
|
#!/bin/sh
|
|
set -eu
|
|
printf 'check_komodo %s\\n' "$1" >> "$CALL_LOG"
|
|
mkdir -p "$1/reports"
|
|
{
|
|
printf 'controller\\t%s\\n' "${KOMODO_CONTROLLER_MODIFIED:-2026-07-27T15:46:17}"
|
|
printf 'machine\\t%s\\n' "${KOMODO_MACHINE_MODIFIED:-2026-07-27T15:39:56}"
|
|
printf 'frame\\t%s\\n' "${KOMODO_FRAME_MODIFIED:-2026-07-03T15:56:30}"
|
|
} > "$1/reports/komodo-product-modified.tsv"
|
|
{
|
|
printf '413804\\timage/png\\thttps://komodostation.com/wp-content/uploads/2025/11/Frame_BG.png\\n'
|
|
printf '456489\\timage/jpeg\\thttps://komodostation.com/wp-content/uploads/2025/11/source_SF_headsetControllers_front_2.jpg\\n'
|
|
if [ -n "${KOMODO_EXTRA_ASSET_URL:-}" ]; then
|
|
printf '999999\\timage/jpeg\\t%s\\n' "$KOMODO_EXTRA_ASSET_URL"
|
|
fi
|
|
} > "$1/reports/komodo-visual-assets.tsv"
|
|
""",
|
|
)
|
|
self._write_python(
|
|
"write_run_summary.py",
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
Path(__import__("os").environ["CALL_LOG"]).open("a").write("write_run_summary " + " ".join(sys.argv[1:]) + "\\n")
|
|
output = Path(sys.argv[sys.argv.index("--output") + 1])
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text("# Summary\\n", encoding="utf-8")
|
|
""",
|
|
)
|
|
|
|
for name in [
|
|
"check_steamkit_pics.sh",
|
|
"check_steamdb.sh",
|
|
"check_steamtracking.sh",
|
|
"check_steamvr_depots.sh",
|
|
"check_steamos_mirror.sh",
|
|
"check_valve_endpoints.sh",
|
|
"check_customs_shipments.sh",
|
|
"save_visual_assets.py",
|
|
"compare_runs.py",
|
|
"write_frame_focus_report.py",
|
|
"draft_status_update.py",
|
|
]:
|
|
if name.endswith(".py"):
|
|
self._write_python(name, f"raise SystemExit('unexpected {name}')\n")
|
|
else:
|
|
self._write_script(
|
|
name,
|
|
f"""
|
|
#!/bin/sh
|
|
echo 'unexpected {name}' >&2
|
|
exit 42
|
|
""",
|
|
)
|
|
|
|
def _write_script(self, name, body):
|
|
path = self.scripts / name
|
|
path.write_text(textwrap.dedent(body).lstrip(), encoding="utf-8")
|
|
path.chmod(0o755)
|
|
|
|
def _write_python(self, name, body):
|
|
path = self.scripts / name
|
|
path.write_text(textwrap.dedent(body).lstrip(), encoding="utf-8")
|
|
path.chmod(0o755)
|
|
|
|
def run_script(self, *args):
|
|
env = os.environ.copy()
|
|
env["CALL_LOG"] = str(self.log)
|
|
return subprocess.run(
|
|
list(args),
|
|
cwd=self.repo,
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
def write_previous_komodo_report(
|
|
self,
|
|
run_date,
|
|
frame_modified,
|
|
controller_modified="2026-07-27T15:46:17",
|
|
machine_modified="2026-07-27T15:39:56",
|
|
):
|
|
report = self.repo / "runs" / run_date / "reports" / "komodo-product-modified.tsv"
|
|
report.parent.mkdir(parents=True)
|
|
report.write_text(
|
|
"\n".join(
|
|
[
|
|
f"controller\t{controller_modified}",
|
|
f"machine\t{machine_modified}",
|
|
f"frame\t{frame_modified}",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def write_previous_komodo_assets(self, run_date, extra_urls=None):
|
|
report = self.repo / "runs" / run_date / "reports" / "komodo-visual-assets.tsv"
|
|
report.parent.mkdir(parents=True, exist_ok=True)
|
|
rows = [
|
|
"413804\timage/png\thttps://komodostation.com/wp-content/uploads/2025/11/Frame_BG.png",
|
|
"456489\timage/jpeg\thttps://komodostation.com/wp-content/uploads/2025/11/source_SF_headsetControllers_front_2.jpg",
|
|
]
|
|
for index, url in enumerate(extra_urls or [], start=1):
|
|
rows.append(f"99999{index}\timage/jpeg\t{url}")
|
|
report.write_text("\n".join(rows + [""]), encoding="utf-8")
|
|
|
|
def test_run_watch_komodo_only_runs_komodo_and_skips_other_sources(self):
|
|
base_dir = self.repo / "runs"
|
|
self.write_previous_komodo_report("2099-01-01", "2026-07-03T15:56:30")
|
|
self.write_previous_komodo_assets("2099-01-01")
|
|
|
|
result = self.run_script(
|
|
str(self.scripts / "run_watch.sh"),
|
|
"--komodo-only",
|
|
"2099-01-02",
|
|
str(base_dir),
|
|
"/tmp/not-used",
|
|
)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
self.assertIn(f"Run dir: {base_dir / '2099-01-02'}", result.stdout)
|
|
self.assertIn("Komodo updated: no (matches previous run 2099-01-01)", result.stdout)
|
|
self.assertIn(
|
|
"Komodo timestamps: controller=2026-07-27T15:46:17, machine=2026-07-27T15:39:56, frame=2026-07-03T15:56:30",
|
|
result.stdout,
|
|
)
|
|
self.assertIn("Komodo new assets: no (2 current, matches previous run 2099-01-01)", result.stdout)
|
|
self.assertIn("Komodo-only run complete.", result.stdout)
|
|
calls = self.log.read_text(encoding="utf-8")
|
|
self.assertIn("check_komodo", calls)
|
|
self.assertIn("write_run_summary", calls)
|
|
self.assertNotIn("unexpected", calls)
|
|
|
|
def test_run_frame_watch_komodo_only_does_not_write_frame_focus_report(self):
|
|
base_dir = self.repo / "runs"
|
|
|
|
result = self.run_script(
|
|
str(self.scripts / "run_frame_watch.sh"),
|
|
"--komodo-only",
|
|
"2099-01-03",
|
|
str(base_dir),
|
|
)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
self.assertIn("Komodo-only run complete.", result.stdout)
|
|
calls = self.log.read_text(encoding="utf-8")
|
|
self.assertIn("check_komodo", calls)
|
|
self.assertNotIn("write_frame_focus_report", calls)
|
|
|
|
def test_run_watch_komodo_only_reports_when_komodo_changed(self):
|
|
base_dir = self.repo / "runs"
|
|
self.write_previous_komodo_report("2099-01-03", "2026-07-03T15:56:30")
|
|
self.write_previous_komodo_assets("2099-01-03")
|
|
os.environ["KOMODO_FRAME_MODIFIED"] = "2026-07-28T10:11:12"
|
|
self.addCleanup(lambda: os.environ.pop("KOMODO_FRAME_MODIFIED", None))
|
|
|
|
result = self.run_script(
|
|
str(self.scripts / "run_watch.sh"),
|
|
"--komodo-only",
|
|
"2099-01-04",
|
|
str(base_dir),
|
|
)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
self.assertIn("Komodo updated: yes (changed since previous run 2099-01-03)", result.stdout)
|
|
self.assertIn("Komodo changes:", result.stdout)
|
|
self.assertIn("- frame: 2026-07-03T15:56:30 -> 2026-07-28T10:11:12", result.stdout)
|
|
self.assertIn(
|
|
"Komodo timestamps: controller=2026-07-27T15:46:17, machine=2026-07-27T15:39:56, frame=2026-07-28T10:11:12",
|
|
result.stdout,
|
|
)
|
|
|
|
def test_run_watch_komodo_only_reports_new_asset_urls(self):
|
|
base_dir = self.repo / "runs"
|
|
self.write_previous_komodo_report("2099-01-05", "2026-07-03T15:56:30")
|
|
self.write_previous_komodo_assets("2099-01-05")
|
|
new_url = "https://komodostation.com/wp-content/uploads/2026/07/new-frame-asset.jpg"
|
|
os.environ["KOMODO_EXTRA_ASSET_URL"] = new_url
|
|
self.addCleanup(lambda: os.environ.pop("KOMODO_EXTRA_ASSET_URL", None))
|
|
|
|
result = self.run_script(
|
|
str(self.scripts / "run_watch.sh"),
|
|
"--komodo-only",
|
|
"2099-01-06",
|
|
str(base_dir),
|
|
)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
self.assertIn("Komodo new assets: yes (1 new, 3 current, since previous run 2099-01-05)", result.stdout)
|
|
self.assertIn("Komodo new asset URLs:", result.stdout)
|
|
self.assertIn(f"- {new_url}", result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|