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