tambo 0.7.0

Install and prove the whole AI knowledge-tool kit
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

#!/usr/bin/env python3
"""Offline controls for the public stack-table drift check."""

import importlib.util
from contextlib import redirect_stderr, redirect_stdout
import io
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch, MagicMock

SCRIPT = Path(__file__).with_name("check-stack-table.py")
CANONICAL = SCRIPT.parents[1] / "docs/stack-table.md"
SPEC = importlib.util.spec_from_file_location("stack_table", SCRIPT)
CHECK = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(CHECK)


class StackTableControls(unittest.TestCase):
    def check_copy(self, data):
        with tempfile.TemporaryDirectory() as directory:
            readme = Path(directory) / "README.md"
            readme.write_bytes(data)
            return subprocess.run(
                [sys.executable, str(SCRIPT), "--canonical", str(CANONICAL), str(readme)],
                capture_output=True, text=True, check=False,
            )

    def test_canonical_copy_passes_but_changed_wording_fails(self):
        data = CANONICAL.read_bytes()
        self.assertEqual(self.check_copy(data).returncode, 0)
        changed = data.replace(b"can read policies from Quipu", b"ignores policies from Quipu")
        self.assertNotEqual(changed, data)
        result = self.check_copy(changed)
        self.assertEqual(result.returncode, 1)
        self.assertIn("differs", result.stderr)

    def test_lead_drift_is_not_normalized_away(self):
        data = CANONICAL.read_bytes()
        changed = data.replace(b"governed, bitemporal", b"unaccountable, timeless")
        self.assertNotEqual(changed, data)
        self.assertEqual(self.check_copy(changed).returncode, 1)

    def test_lead_without_table_or_separator_fails(self):
        for body in (b"An overview.\n", b"An overview.\n\n| tool | value |\n| bad | row |\n"):
            with self.subTest(body=body):
                with self.assertRaisesRegex(ValueError, "complete stack overview"):
                    CHECK.marked_table(CHECK.START + b"\n" + body + CHECK.END + b"\n", "fixture")

    def test_missing_duplicate_or_reversed_markers_fail(self):
        data = CANONICAL.read_bytes()
        for bad in (data.replace(CHECK.START, b""), data + CHECK.START,
                    CHECK.END + b"\n" + CHECK.START + b"\n"):
            with self.subTest(data=bad[:50]):
                self.assertEqual(self.check_copy(bad).returncode, 1)

    def test_whitespace_drift_is_not_normalized_away(self):
        data = CANONICAL.read_bytes().replace(b"| tool |", b"|  tool |")
        self.assertEqual(self.check_copy(data).returncode, 1)

    def test_remote_error_is_not_a_clean_zero(self):
        with patch.object(CHECK, "remote_readme", side_effect=OSError("offline")), \
             patch.object(sys, "argv", [str(SCRIPT), "--remote"]), \
             redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
            self.assertEqual(CHECK.main(), 1)

    def test_new_rikuy_and_shantytown_missing_blocks_fail(self):
        canonical = CANONICAL.read_bytes()
        for missing, expected in ((None, 0), ("shantytown", 1), ("rikuy", 1)):
            seen = []

            def read(repo):
                seen.append(repo)
                return b"A README without a stack block.\n" if repo == missing else canonical

            with self.subTest(missing=missing), \
                 patch.object(CHECK, "remote_readme", side_effect=read), \
                 patch.object(sys, "argv", [str(SCRIPT), "--remote"]), \
                 redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
                self.assertEqual(CHECK.main(), expected)
                self.assertEqual(seen.count("shantytown"), 1)
                self.assertEqual(seen.count("rikuy"), 1)
                self.assertEqual(len(seen), 12)

    def test_remote_response_has_a_size_limit(self):
        response = MagicMock()
        response.__enter__.return_value.read.return_value = b"x" * (CHECK.MAX_BYTES + 1)
        with patch.object(CHECK, "urlopen", return_value=response):
            with self.assertRaisesRegex(ValueError, "size limit"):
                CHECK.remote_readme("tambo")


if __name__ == "__main__":
    unittest.main()