#!/usr/bin/env python3 """Run: python3 -m unittest discover -s tests -p 'test_dashboard.py' -v""" from __future__ import annotations import importlib.util import json import os import tempfile import threading import unittest from datetime import datetime, timezone from http.server import ThreadingHTTPServer from pathlib import Path from urllib.error import HTTPError from urllib.request import Request, urlopen ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location("dashboard_app", ROOT / "dashboard" / "app.py") app = importlib.util.module_from_spec(spec) assert spec and spec.loader spec.loader.exec_module(app) class RepoFixture: def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.repo = Path(self.temp.name) / "repository" self.r1 = self.repo / "releases" / "r1" self.r1.mkdir(parents=True) (self.repo / "current").symlink_to("releases/r1") def make_file(self, name, *, contents="package-bytes", timestamp=1735853760): path = self.r1 / name path.parent.mkdir(parents=True,exist_ok=True) path.write_text(contents,encoding="utf-8") os.utime(path,(timestamp,timestamp)) return path def make_catalog(self): self.make_file("apt/pool/r/rustproxy/rustproxy_0.34.6-2_amd64.deb") self.make_file("apt/pool/r/rustproxy/rustproxy-docker-agent_0.34.6-2_amd64.deb") self.make_file("rpm/x86_64/rustproxy-0.34.6-2.x86_64.rpm") self.make_file("rpm/x86_64/rustproxy-docker-agent-0.34.6-2.x86_64.rpm") self.make_file("opnsense/26.7/amd64/os-rustproxy-0.34.6_123.pkg") self.make_file("opnsense/packagesite.pkg",contents="index") self.make_file("opnsense/os-rustproxy-latest.pkg",contents="duplicate") self.make_file("freebsd/FreeBSD:15:amd64/rustproxy-0.34.6_123.pkg") self.make_file("freebsd/rustproxy-latest.pkg",contents="duplicate") self.make_file("LINUX-BUILD-INFO.txt",contents="app_version=0.34.6\npackage_release=2\nsource_commit=abc123\nbuilt_at=2026-10-22T08:41:00Z\n") self.make_file("opnsense/BUILD-INFO.txt",contents="app_version=0.34.6\npackage_version=0.34.6_123\nsource_commit=abc123\nbuilt_at=2026-10-22T08:43:00Z\n") self.make_file("freebsd/BUILD-INFO.txt",contents="app_version=0.34.6\npackage_version=0.34.6_123\nsource_commit=abc123\nbuilt_at=2026-10-22T08:45:00Z\n") class CatalogTests(RepoFixture, unittest.TestCase): def test_scans_real_file_layout_without_index_duplicates(self): self.make_catalog() catalog = app.generate_catalog(self.repo) self.assertEqual(6, catalog["summary"]["packages_count"]) self.assertEqual(3, catalog["summary"]["channels_published"]) self.assertEqual(4, catalog["summary"]["channel_counts"]["linux"]) self.assertEqual(1, catalog["summary"]["channel_counts"]["opnsense"]) self.assertEqual(1, catalog["summary"]["channel_counts"]["freebsd"]) self.assertEqual("0.34.6-2",catalog["builds"]["linux"]["package_version"]) self.assertEqual("2026-10-22T08:45:00Z",catalog["summary"]["latest_build_at"]) linux = [p for p in catalog["packages"] if p["format"] == "RPM"] self.assertEqual({"rustproxy","rustproxy-docker-agent"}, {p["name"] for p in linux}) self.assertEqual("2026-10-22T08:41:00Z", linux[0]["built_at"]) self.assertEqual("build",linux[0]["date_source"]) opnsense = next(p for p in catalog["packages"] if p["channel"] == "opnsense") self.assertTrue(opnsense["download_url"].endswith("/26.7/amd64/os-rustproxy-0.34.6_123.pkg")) freebsd = next(p for p in catalog["packages"] if p["channel"] == "freebsd") self.assertIn("/freebsd/FreeBSD%3A15%3Aamd64/",freebsd["download_url"]) self.assertEqual("2026-10-22T08:45:00Z",freebsd["built_at"]) def test_mtime_backwards_compatibility(self): self.make_file("LINUX-BUILD-INFO.txt",contents="app_version=0.21.0\npackage_release=5\n") self.make_file("rpm/x86_64/rustproxy-0.21.0-5.x86_64.rpm") catalog = app.generate_catalog(self.repo) self.assertEqual("file", catalog["builds"]["linux"]["date_source"]) self.assertEqual("file", catalog["packages"][0]["date_source"]) self.assertEqual("0.21.0-5",catalog["packages"][0]["version"]) self.assertTrue(catalog["packages"][0]["built_at"].endswith("Z")) def test_symlink_rotation_gives_current_packages(self): self.make_file("rpm/x86_64/rustproxy-0.1.0-1.x86_64.rpm") v2 = self.repo / "releases" / "r2" v2.mkdir() (v2 / "rpm/x86_64").mkdir(parents=True) (v2 / "rpm/x86_64/rustproxy-0.2.0-1.x86_64.rpm").write_bytes(b"new package") before = app.generate_catalog(self.repo) self.assertEqual("0.1.0-1",before["packages"][0]["version"]) (self.repo / "current.new").symlink_to("releases/r2") os.replace(self.repo / "current.new",self.repo / "current") after = app.generate_catalog(self.repo) self.assertEqual("0.2.0-1",after["packages"][0]["version"]) def test_empty_repo_and_untrusted_symlink(self): first = app.generate_catalog(self.repo) self.assertFalse(first["ready"]) self.assertEqual([],first["packages"]) (self.repo / "current").unlink() (self.repo / "current").symlink_to(self.temp.name) # outside repo root second = app.generate_catalog(self.repo) self.assertFalse(second["ready"]) self.assertEqual([],second["packages"]) def test_regexes_and_metadata_parser(self): self.assertEqual(("rustproxy-docker-agent","0.34.6-2","x86_64"),app.parse_filename("rustproxy-docker-agent-0.34.6-2.x86_64.rpm","rpm")) self.assertEqual(("os-rustproxy","0.34.6_123","amd64"),app.parse_filename("os-rustproxy-0.34.6_123.pkg","pkg")) self.assertIsNone(app.parse_filename("packagesite.pkg","pkg")) self.assertIsNone(app.parse_filename("rustproxy-latest.pkg","pkg")) self.assertIsNone(app.parse_filename("repomd.xml","rpm")) self.assertEqual({},app.parse_metadata(self.repo/"missing-info.txt")) class HttpTests(RepoFixture, unittest.TestCase): def setUp(self): super().setUp() self.make_catalog() original_cache = app.CACHE app.CACHE = app.CatalogCache(self.repo,ttl=1) self.addCleanup(lambda: setattr(app,"CACHE",original_cache)) self.server = ThreadingHTTPServer(("127.0.0.1",0),app.DashboardHandler) self.thread = threading.Thread(target=self.server.serve_forever,daemon=True) self.thread.start() self.addCleanup(self.server.server_close) self.addCleanup(self.server.shutdown) self.base = f"http://127.0.0.1:{self.server.server_port}" def test_http_catalog_page_assets_head(self): with urlopen(self.base + "/api/v1/packages") as response: self.assertIn("no-store",response.headers["Cache-Control"]) catalog = json.load(response) self.assertEqual(6,len(catalog["packages"])) request = Request(self.base + "/",headers={"X-Repo-Site":"opnsense"}) with urlopen(request) as response: html = response.read().decode() self.assertIn('data-site="opnsense"',html) self.assertIn("Package Registry",html) self.assertIn("Content-Security-Policy",response.headers) with urlopen(self.base + "/freebsd/") as response: self.assertIn('data-site="freebsd"',response.read().decode()) with urlopen(self.base + "/ui/style.css") as response: self.assertIn("text/css",response.headers["Content-Type"]) self.assertIn("--accent:#f69b74",response.read().decode()) request = Request(self.base + "/ui/app.js", method="HEAD") with urlopen(request) as response: self.assertEqual(200,response.status) self.assertEqual(b"",response.read()) with self.assertRaises(HTTPError) as raised: urlopen(self.base + "/../something") self.assertEqual(404,raised.exception.code) if __name__ == "__main__": unittest.main()