Files
computer-playbook/tests/unit/python/utils/cache/test_files.py
T
kevinveenbirkenbachandClaude Opus 5.5 c9e7f950d9 Merge branch 'feature/updater' into feature/store-i18n
68 conflicts, resolved per hunk; resolutions agreed with the Updater
session were applied as agreed:

- svc-db-postgres (superuser password script and core) and
  sys-svc-rdbms realign_dedicated_db: updater's argv/stdin redesign with
  expand_argument_vars: false, nested into store-i18n's block structure.
- svc-db-redis 01_init: store-i18n's ACL password on stdin.
- sys-svc-proxy forwarded.conf.j2: store-i18n's handover-aware file,
  without the forwarded_base include (Host and X-Forwarded-Proto would
  be sent twice); forwarded_base serves the inline consumers, e.g.
  web-app-seaweedfs nginx.conf.j2.
- basic.conf.j2 and web-app-openproject proxy.conf.j2: issuer redirect,
  then handover, else updater's location/sso_acl.conf.j2; the
  server-level webserver_extra_configuration hook stays.
- web-app-keycloak: updater's admin-REST settings step and renumbered
  update steps; update/01_client.yml removed; store-i18n's design step
  and LDAP block kept; a73435cc9d arrived through rename detection.
- web-app-nextcloud OpenProject provisioning: updater's rewrite, all
  credentials on the Ruby runner's stdin.
- web-app-semaphore: store-i18n's canonical web root (a1d7835863).
- Role READMEs: no generated Cosmos/Architecture/Quick Setup/Credits
  sections; desk-* paths resolved to the dsk-* renames.

Integration fixes for what merged cleanly but did not fit together:

- web-app-matrix ansible vars: drop updater's second
  matrix_client_element_configuration_extension_json, which silently
  overrode store-i18n's design-aware value.
- url.origin instead of a stripped url.base in 19 places store-i18n
  added since 44aaa68924, same lookup targets (the double-slash lint).
- Package batch (18ea75049e, cb4d340de9): installs the deploy-start
  batch covers are removed (caffeine, cursor, blender, npm in dsk-gnt-*,
  with shared references), the rest carry batch: false and a nocheck
  with the verified reason (PyPI fallback, before the batch, vendor
  repository, configured flavor only, addon gate); dev-npm references
  the root npm id.
- sys-token-store write_many.yml: store-i18n's change-gated block and
  IS_SWARM_MODE.
- web-app-openproject: IS_COMPOSE_MODE in 02_migrate/run.yml; the
  seeded-admin lock (ccbf8f168d) ported into 04_admin/provision.yml and
  covered by the admin-state digest.
- Bucket policy (3b4ac40df5 and follow-ups) moved to web-svc-seaweedfs,
  where store-i18n's object store lives; the S3 admin secret travels on
  stdin instead of a task-level environment.
- act workflow.sh: no Exception-labelled comment.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-29 02:28:16 +01:00

276 lines
11 KiB
Python

"""Unit tests for `utils.cache.files`.
Covers the project-tree walk cache, the file-content cache, and the
combined ``iter_project_files_with_content`` helper. All tests run
against a synthetic tempdir; ``PROJECT_ROOT`` is monkey-patched onto
the module so the real repo tree is never touched.
"""
from __future__ import annotations
import subprocess
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from types import SimpleNamespace
from unittest.mock import patch
from utils.cache import files as files_module
from utils.cache.files import (
_reset,
iter_project_files,
iter_project_files_with_content,
read_text,
)
from utils.roles.mapping import ROLE_FILE_META_SERVICES, ROLE_FILE_TASKS_MAIN
def _touch(path: Path, content: str = "") -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return path
class _ProjectRootFixture:
"""Mixin that sets up a synthetic PROJECT_ROOT for every test in the
subclass. Combined with ``unittest.TestCase`` via multiple inheritance
so the lint guard ``test_all_test_files_have_tests`` can see the
``TestCase`` base directly on the concrete test classes.
"""
def setUp(self) -> None:
super().setUp()
_reset()
self._tmp = TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.root = Path(self._tmp.name)
self._patcher = patch.object(files_module, "PROJECT_ROOT", self.root)
self._patcher.start()
self.addCleanup(self._patcher.stop)
self.addCleanup(_reset)
class TestReadText(_ProjectRootFixture, unittest.TestCase):
def test_returns_utf8_content(self) -> None:
path = _touch(self.root / "a.txt", "héllo")
self.assertEqual(read_text(str(path)), "héllo")
def test_second_call_hits_cache(self) -> None:
path = _touch(self.root / "a.txt", "first")
self.assertEqual(read_text(str(path)), "first")
path.write_text("changed", encoding="utf-8")
self.assertEqual(read_text(str(path)), "first")
def test_reset_invalidates_cache(self) -> None:
path = _touch(self.root / "a.txt", "first")
self.assertEqual(read_text(str(path)), "first")
path.write_text("second", encoding="utf-8")
_reset()
self.assertEqual(read_text(str(path)), "second")
def test_missing_file_raises(self) -> None:
with self.assertRaises(FileNotFoundError):
read_text(str(self.root / "nope.txt"))
class TestIterProjectFiles(_ProjectRootFixture, unittest.TestCase):
def setUp(self) -> None:
super().setUp()
_touch(self.root / f"roles/web-app-foo/{ROLE_FILE_META_SERVICES}", "foo: 1")
_touch(self.root / "roles/web-app-foo/templates/x.j2", "{{ x }}")
_touch(self.root / "docs/readme.md", "# Hi")
_touch(self.root / "tests/unit/test_a.py", "import unittest")
_touch(self.root / ".git/HEAD", "ref: refs/heads/main")
_touch(self.root / "node_modules/pkg/index.js", "module.exports = {};")
_touch(self.root / "vendor/nette/utils/readme.md", "# vendored")
_touch(self.root / "__pycache__/dead.pyc", "")
_touch(self.root / "build/ruff-cache-0/0.16.7/entry.json", "{}")
_touch(self.root / "cli/build/docs/readme/overview.py", "x = 1")
def test_skip_dirs_pruned(self) -> None:
all_paths = set(iter_project_files())
self.assertFalse(
any("/.git/" in p for p in all_paths),
f".git not pruned in {all_paths}",
)
self.assertFalse(any("/__pycache__/" in p for p in all_paths))
self.assertFalse(any("/node_modules/" in p for p in all_paths))
self.assertFalse(any("/vendor/" in p for p in all_paths))
def test_build_is_pruned_at_the_root_only(self) -> None:
all_paths = set(iter_project_files())
self.assertFalse(
any(p.startswith(f"{self.root}/build/") for p in all_paths),
f"root build/ artefacts reached the walk: {all_paths}",
)
self.assertTrue(
any(p.endswith("cli/build/docs/readme/overview.py") for p in all_paths),
f"cli/build is source and must survive the prune: {all_paths}",
)
def test_git_local_exclude_dirs_pruned(self) -> None:
"""Plain directory entries in .git/info/exclude are pruned; file
patterns and globs are left to git."""
_touch(
self.root / ".git/info/exclude",
"# comment\n/local-tool-out/\nsettings.local.json\n*.tmp\n",
)
_touch(self.root / "local-tool-out/report.md", "generated")
_reset()
all_paths = set(iter_project_files())
self.assertFalse(any("/local-tool-out/" in p for p in all_paths))
self.assertTrue(any(p.endswith("readme.md") for p in all_paths))
def test_extensions_filter(self) -> None:
ymls = list(iter_project_files(extensions=(".yml",)))
self.assertEqual(len(ymls), 1)
self.assertTrue(ymls[0].endswith("services.yml"))
markdown = list(iter_project_files(extensions=(".md",)))
self.assertEqual(len(markdown), 1)
self.assertTrue(markdown[0].endswith("readme.md"))
def test_exclude_tests_skips_tests_subtree(self) -> None:
with_tests = list(iter_project_files(extensions=(".py",)))
self.assertEqual(len(with_tests), 2)
without_tests = list(
iter_project_files(extensions=(".py",), exclude_tests=True)
)
self.assertEqual(len(without_tests), 1)
self.assertTrue(without_tests[0].endswith("cli/build/docs/readme/overview.py"))
def test_exclude_dirs_path_segment_match(self) -> None:
kept = list(iter_project_files(exclude_dirs=("docs",)))
self.assertFalse(any(p.endswith("readme.md") for p in kept))
def test_walk_is_cached_across_calls(self) -> None:
first = set(iter_project_files())
_touch(self.root / "after.txt", "late")
second = set(iter_project_files())
self.assertEqual(first, second)
_reset()
third = set(iter_project_files())
self.assertIn(str(self.root / "after.txt"), third)
class TestIterProjectFilesWithContent(_ProjectRootFixture, unittest.TestCase):
def test_yields_path_and_content(self) -> None:
_touch(self.root / "a.txt", "alpha")
_touch(self.root / "b.txt", "beta")
result = sorted(iter_project_files_with_content(extensions=(".txt",)))
self.assertEqual(
result,
sorted(
[
(str(self.root / "a.txt"), "alpha"),
(str(self.root / "b.txt"), "beta"),
]
),
)
def test_non_utf8_files_silently_skipped(self) -> None:
good = self.root / "good.bin"
bad = self.root / "bad.bin"
good.write_text("ok", encoding="utf-8")
bad.write_bytes(b"\xff\xfe\x00invalid")
result = list(iter_project_files_with_content(extensions=(".bin",)))
self.assertEqual(result, [(str(good), "ok")])
class TestReset(_ProjectRootFixture, unittest.TestCase):
def test_reset_clears_both_caches(self) -> None:
_touch(self.root / "x.txt", "v1")
list(iter_project_files())
read_text(str(self.root / "x.txt"))
_touch(self.root / "y.txt", "v2")
(self.root / "x.txt").write_text("v1-changed", encoding="utf-8")
self.assertNotIn(str(self.root / "y.txt"), set(iter_project_files()))
self.assertEqual(read_text(str(self.root / "x.txt")), "v1")
_reset()
self.assertIn(str(self.root / "y.txt"), set(iter_project_files()))
self.assertEqual(read_text(str(self.root / "x.txt")), "v1-changed")
class TestRepositoryFiles(_ProjectRootFixture, unittest.TestCase):
TRACKED = f"roles/web-app-nextcloud/{ROLE_FILE_TASKS_MAIN}"
CUSTOM = f"roles/web-app-cst-demo/{ROLE_FILE_TASKS_MAIN}"
def _git(self, *listings: bytes):
return patch.object(
files_module.subprocess,
"run",
side_effect=[SimpleNamespace(stdout=listing) for listing in listings],
)
def test_lists_placed_custom_roles_next_to_the_tracked_files(self) -> None:
_touch(self.root / self.TRACKED, "- y")
_touch(self.root / self.CUSTOM, "- x")
_touch(self.root / "roles/web-app-cst-demo/__pycache__/x.pyc", "")
with self._git(f"{self.TRACKED}\0".encode()):
self.assertEqual(
(self.CUSTOM, self.TRACKED),
files_module.repository_files(),
)
def test_lists_only_the_tracked_files_without_placed_custom_roles(self) -> None:
_touch(self.root / self.TRACKED, "- y")
_touch(self.root / "build/install.stamp", "")
with self._git(f"{self.TRACKED}\0".encode()):
self.assertEqual((self.TRACKED,), files_module.repository_files())
def test_untracked_adds_the_files_gitignore_does_not_exclude(self) -> None:
with self._git(f"{self.TRACKED}\0".encode(), b"new.py\0") as run:
self.assertEqual(
("new.py", self.TRACKED),
files_module.repository_files(untracked=True),
)
self.assertEqual(
["--others", "--exclude-standard"],
run.call_args_list[1].args[0][-2:],
)
def test_a_git_failure_propagates(self) -> None:
with (
patch.object(
files_module.subprocess,
"run",
side_effect=subprocess.CalledProcessError(128, "git"),
),
self.assertRaises(subprocess.CalledProcessError),
):
files_module.repository_files()
class TestWalkIsKeyedOnTheRoot(_ProjectRootFixture, unittest.TestCase):
def test_a_second_root_is_walked_rather_than_answered_from_the_first(
self,
) -> None:
_touch(self.root / "first.txt", "a")
self.assertEqual(
{str(self.root / "first.txt")},
set(iter_project_files()),
)
with TemporaryDirectory() as other:
other_root = Path(other)
_touch(other_root / "second.txt", "b")
with patch.object(files_module, "PROJECT_ROOT", other_root):
self.assertEqual(
{str(other_root / "second.txt")},
set(iter_project_files()),
"the walk answered for the previous root, so a test that "
"points PROJECT_ROOT at a fixture silently sweeps the real "
"repository instead",
)
self.assertEqual(
{str(self.root / "first.txt")},
set(iter_project_files()),
)
if __name__ == "__main__": # pragma: no cover
unittest.main()