repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources15d ago| #1 | """ |
| #2 | Phase 5: Memory Bank Isolation Tests |
| #3 | |
| #4 | Validates: |
| #5 | 1. BankManager.create_bank() creates isolated directories |
| #6 | 2. BankManager.list_banks() returns all banks including default |
| #7 | 3. BankManager.delete_bank() removes banks |
| #8 | 4. BankManager.get_bank_db_path() returns correct paths |
| #9 | 5. Mnemosyne(bank="work") uses isolated database |
| #10 | 6. Module-level set_bank() switches global default |
| #11 | 7. Data written to one bank is invisible to another |
| #12 | 8. Bank names are validated (alphanumeric + hyphens/underscores) |
| #13 | 9. Default bank cannot be deleted without force |
| #14 | """ |
| #15 | |
| #16 | import os |
| #17 | import sys |
| #18 | import pytest |
| #19 | import tempfile |
| #20 | import sqlite3 |
| #21 | from pathlib import Path |
| #22 | |
| #23 | sys.path.insert(0, str(Path(__file__).parent.parent)) |
| #24 | |
| #25 | from mnemosyne.core.banks import BankManager, create_bank, delete_bank, list_banks, bank_exists |
| #26 | from mnemosyne.core.memory import Mnemosyne, set_bank, get_bank, remember, recall, get_stats |
| #27 | |
| #28 | |
| #29 | # ============================================================================ |
| #30 | # BankManager unit tests |
| #31 | # ============================================================================ |
| #32 | |
| #33 | class TestBankManager: |
| #34 | """Unit tests for BankManager class.""" |
| #35 | |
| #36 | def test_create_bank_creates_directory(self): |
| #37 | """Creating a bank should create a directory with a DB file.""" |
| #38 | with tempfile.TemporaryDirectory() as tmpdir: |
| #39 | data_dir = Path(tmpdir) |
| #40 | mgr = BankManager(data_dir) |
| #41 | db_path = mgr.create_bank("work") |
| #42 | assert db_path.exists() |
| #43 | assert db_path.parent.name == "work" |
| #44 | assert db_path.name == "mnemosyne.db" |
| #45 | |
| #46 | def test_create_bank_duplicate_raises(self): |
| #47 | """Creating a duplicate bank should raise ValueError.""" |
| #48 | with tempfile.TemporaryDirectory() as tmpdir: |
| #49 | mgr = BankManager(Path(tmpdir)) |
| #50 | mgr.create_bank("work") |
| #51 | with pytest.raises(ValueError, match="already exists"): |
| #52 | mgr.create_bank("work") |
| #53 | |
| #54 | def test_create_bank_invalid_name_raises(self): |
| #55 | """Invalid bank names should raise ValueError.""" |
| #56 | with tempfile.TemporaryDirectory() as tmpdir: |
| #57 | mgr = BankManager(Path(tmpdir)) |
| #58 | with pytest.raises(ValueError): |
| #59 | mgr.create_bank("bank with spaces") |
| #60 | with pytest.raises(ValueError): |
| #61 | mgr.create_bank("bank/with/slashes") |
| #62 | with pytest.raises(ValueError): |
| #63 | mgr.create_bank("bank.with.dots") |
| #64 | |
| #65 | def test_list_banks_includes_default(self): |
| #66 | """list_banks() should always include 'default'.""" |
| #67 | with tempfile.TemporaryDirectory() as tmpdir: |
| #68 | mgr = BankManager(Path(tmpdir)) |
| #69 | banks = mgr.list_banks() |
| #70 | assert "default" in banks |
| #71 | |
| #72 | def test_list_banks_after_create(self): |
| #73 | """list_banks() should include newly created banks.""" |
| #74 | with tempfile.TemporaryDirectory() as tmpdir: |
| #75 | mgr = BankManager(Path(tmpdir)) |
| #76 | mgr.create_bank("work") |
| #77 | mgr.create_bank("personal") |
| #78 | banks = mgr.list_banks() |
| #79 | assert "work" in banks |
| #80 | assert "personal" in banks |
| #81 | assert "default" in banks |
| #82 | |
| #83 | def test_delete_bank_removes_directory(self): |
| #84 | """delete_bank() should remove the bank directory.""" |
| #85 | with tempfile.TemporaryDirectory() as tmpdir: |
| #86 | mgr = BankManager(Path(tmpdir)) |
| #87 | mgr.create_bank("temp") |
| #88 | assert mgr.bank_exists("temp") |
| #89 | mgr.delete_bank("temp") |
| #90 | assert not mgr.bank_exists("temp") |
| #91 | |
| #92 | def test_delete_nonexistent_bank_returns_false(self): |
| #93 | """delete_bank() on non-existent bank returns False.""" |
| #94 | with tempfile.TemporaryDirectory() as tmpdir: |
| #95 | mgr = BankManager(Path(tmpdir)) |
| #96 | result = mgr.delete_bank("nonexistent") |
| #97 | assert result is False |
| #98 | |
| #99 | def test_delete_default_bank_raises(self): |
| #100 | """Deleting 'default' without force=True should raise.""" |
| #101 | with tempfile.TemporaryDirectory() as tmpdir: |
| #102 | mgr = BankManager(Path(tmpdir)) |
| #103 | with pytest.raises(ValueError, match="force=True"): |
| #104 | mgr.delete_bank("default") |
| #105 | |
| #106 | def test_delete_default_with_force_succeeds(self): |
| #107 | """Deleting 'default' with force=True should succeed.""" |
| #108 | with tempfile.TemporaryDirectory() as tmpdir: |
| #109 | mgr = BankManager(Path(tmpdir)) |
| #110 | # default doesn't have a directory, but force should not raise |
| #111 | result = mgr.delete_bank("default", force=True) |
| #112 | # Returns False because default dir doesn't exist |
| #113 | assert result is False |
| #114 | |
| #115 | def test_bank_exists(self): |
| #116 | """bank_exists() should return True for existing banks.""" |
| #117 | with tempfile.TemporaryDirectory() as tmpdir: |
| #118 | mgr = BankManager(Path(tmpdir)) |
| #119 | assert mgr.bank_exists("default") is True |
| #120 | mgr.create_bank("test") |
| #121 | assert mgr.bank_exists("test") is True |
| #122 | assert mgr.bank_exists("nonexistent") is False |
| #123 | |
| #124 | def test_get_bank_db_path_default(self): |
| #125 | """Default bank should use data_dir/mnemosyne.db.""" |
| #126 | with tempfile.TemporaryDirectory() as tmpdir: |
| #127 | data_dir = Path(tmpdir) |
| #128 | mgr = BankManager(data_dir) |
| #129 | path = mgr.get_bank_db_path("default") |
| #130 | assert path == data_dir / "mnemosyne.db" |
| #131 | |
| #132 | def test_get_bank_db_path_custom(self): |
| #133 | """Custom banks should use banks_dir/<name>/mnemosyne.db.""" |
| #134 | with tempfile.TemporaryDirectory() as tmpdir: |
| #135 | data_dir = Path(tmpdir) |
| #136 | mgr = BankManager(data_dir) |
| #137 | mgr.create_bank("work") |
| #138 | path = mgr.get_bank_db_path("work") |
| #139 | assert path == data_dir / "banks" / "work" / "mnemosyne.db" |
| #140 | |
| #141 | def test_rename_bank(self): |
| #142 | """rename_bank() should move the bank directory.""" |
| #143 | with tempfile.TemporaryDirectory() as tmpdir: |
| #144 | mgr = BankManager(Path(tmpdir)) |
| #145 | mgr.create_bank("old_name") |
| #146 | new_path = mgr.rename_bank("old_name", "new_name") |
| #147 | assert new_path.exists() |
| #148 | assert not mgr.bank_exists("old_name") |
| #149 | assert mgr.bank_exists("new_name") |
| #150 | |
| #151 | def test_rename_default_raises(self): |
| #152 | """Cannot rename 'default' bank.""" |
| #153 | with tempfile.TemporaryDirectory() as tmpdir: |
| #154 | mgr = BankManager(Path(tmpdir)) |
| #155 | with pytest.raises(ValueError, match="Cannot rename"): |
| #156 | mgr.rename_bank("default", "new_name") |
| #157 | |
| #158 | def test_get_bank_stats(self): |
| #159 | """get_bank_stats() should return correct stats.""" |
| #160 | with tempfile.TemporaryDirectory() as tmpdir: |
| #161 | mgr = BankManager(Path(tmpdir)) |
| #162 | mgr.create_bank("stats_test") |
| #163 | stats = mgr.get_bank_stats("stats_test") |
| #164 | assert stats["name"] == "stats_test" |
| #165 | assert stats["exists"] is True |
| #166 | assert stats["db_size_bytes"] >= 0 |
| #167 | |
| #168 | def test_module_level_functions(self): |
| #169 | """Module-level convenience functions should work.""" |
| #170 | with tempfile.TemporaryDirectory() as tmpdir: |
| #171 | data_dir = Path(tmpdir) |
| #172 | db_path = create_bank("mod_test", data_dir) |
| #173 | assert db_path.exists() |
| #174 | assert bank_exists("mod_test", data_dir) |
| #175 | banks = list_banks(data_dir) |
| #176 | assert "mod_test" in banks |
| #177 | delete_bank("mod_test", data_dir) |
| #178 | assert not bank_exists("mod_test", data_dir) |
| #179 | |
| #180 | |
| #181 | # ============================================================================ |
| #182 | # Mnemosyne bank integration tests |
| #183 | # ============================================================================ |
| #184 | |
| #185 | class TestMnemosyneBankIsolation: |
| #186 | """Integration tests verifying Mnemosyne uses correct DB per bank.""" |
| #187 | |
| #188 | def test_mnemosyne_with_bank_uses_isolated_db(self): |
| #189 | """Mnemosyne(bank='work') should use a separate database.""" |
| #190 | with tempfile.TemporaryDirectory() as tmpdir: |
| #191 | data_dir = Path(tmpdir) |
| #192 | os.environ["MNEMOSYNE_DATA_DIR"] = str(data_dir) |
| #193 | try: |
| #194 | mgr = BankManager(data_dir) |
| #195 | mgr.create_bank("work") |
| #196 | |
| #197 | mn_default = Mnemosyne(bank="default") |
| #198 | mn_work = Mnemosyne(bank="work") |
| #199 | |
| #200 | # Should have different DB paths |
| #201 | assert mn_default.db_path != mn_work.db_path |
| #202 | assert "banks" in str(mn_work.db_path) |
| #203 | assert "work" in str(mn_work.db_path) |
| #204 | finally: |
| #205 | del os.environ["MNEMOSYNE_DATA_DIR"] |
| #206 | |
| #207 | def test_data_isolation_between_banks(self): |
| #208 | """Data written to one bank should not appear in another.""" |
| #209 | with tempfile.TemporaryDirectory() as tmpdir: |
| #210 | data_dir = Path(tmpdir) |
| #211 | os.environ["MNEMOSYNE_DATA_DIR"] = str(data_dir) |
| #212 | try: |
| #213 | mgr = BankManager(data_dir) |
| #214 | mgr.create_bank("personal") |
| #215 | |
| #216 | mn_default = Mnemosyne(bank="default") |
| #217 | mn_personal = Mnemosyne(bank="personal") |
| #218 | |
| #219 | # Write to default (unique content to avoid ID collisions) |
| #220 | import time |
| #221 | mn_default.remember("Default bank memory " + str(time.time()), importance=0.8) |
| #222 | time.sleep(0.01) |
| #223 | # Write to personal |
| #224 | mn_personal.remember("Personal bank memory " + str(time.time()), importance=0.9) |
| #225 | |
| #226 | # Query default - should only find default memory |
| #227 | default_results = mn_default.recall("Default bank", top_k=5) |
| #228 | assert any("Default" in r["content"] for r in default_results) |
| #229 | assert not any("Personal" in r["content"] for r in default_results) |
| #230 | |
| #231 | # Query personal - should only find personal memory |
| #232 | personal_results = mn_personal.recall("Personal bank", top_k=5) |
| #233 | assert any("Personal" in r["content"] for r in personal_results) |
| #234 | assert not any("Default" in r["content"] for r in personal_results) |
| #235 | finally: |
| #236 | del os.environ["MNEMOSYNE_DATA_DIR"] |
| #237 | |
| #238 | def test_bank_stats_are_isolated(self): |
| #239 | """Stats should reflect only the current bank.""" |
| #240 | with tempfile.TemporaryDirectory() as tmpdir: |
| #241 | data_dir = Path(tmpdir) |
| #242 | os.environ["MNEMOSYNE_DATA_DIR"] = str(data_dir) |
| #243 | try: |
| #244 | mgr = BankManager(data_dir) |
| #245 | mgr.create_bank("project_a") |
| #246 | |
| #247 | mn_a = Mnemosyne(bank="project_a") |
| #248 | mn_default = Mnemosyne(bank="default") |
| #249 | |
| #250 | import time |
| #251 | mn_a.remember("Project A task " + str(time.time()), importance=0.5) |
| #252 | time.sleep(0.01) |
| #253 | mn_a.remember("Project A meeting " + str(time.time()), importance=0.6) |
| #254 | time.sleep(0.01) |
| #255 | mn_default.remember("General note " + str(time.time()), importance=0.4) |
| #256 | |
| #257 | a_stats = mn_a.get_stats() |
| #258 | default_stats = mn_default.get_stats() |
| #259 | |
| #260 | # Project A should have at least 2 working memories |
| #261 | assert a_stats["beam"]["working_memory"]["total"] >= 2 |
| #262 | # Default should have at least 1 |
| #263 | assert default_stats["beam"]["working_memory"]["total"] >= 1 |
| #264 | finally: |
| #265 | del os.environ["MNEMOSYNE_DATA_DIR"] |
| #266 | |
| #267 | |
| #268 | # ============================================================================ |
| #269 | # Module-level bank switching tests |
| #270 | # ============================================================================ |
| #271 | |
| #272 | class TestModuleLevelBankSwitching: |
| #273 | """Tests for set_bank() and per-call bank parameter.""" |
| #274 | |
| #275 | def test_set_bank_switches_global_default(self): |
| #276 | """set_bank() should change the bank for subsequent module calls.""" |
| #277 | with tempfile.TemporaryDirectory() as tmpdir: |
| #278 | data_dir = Path(tmpdir) |
| #279 | os.environ["MNEMOSYNE_DATA_DIR"] = str(data_dir) |
| #280 | try: |
| #281 | mgr = BankManager(data_dir) |
| #282 | mgr.create_bank("switch_test") |
| #283 | |
| #284 | # Use Mnemosyne class directly for clean isolation |
| #285 | mn_default = Mnemosyne(bank="default") |
| #286 | mn_switch = Mnemosyne(bank="switch_test") |
| #287 | |
| #288 | import time |
| #289 | # Write to default |
| #290 | mn_default.remember("Default memory " + str(time.time()), importance=0.8) |
| #291 | time.sleep(0.01) |
| #292 | # Write to switch bank |
| #293 | mn_switch.remember("Switched memory " + str(time.time()), importance=0.9) |
| #294 | |
| #295 | # Verify isolation via class instances |
| #296 | default_results = mn_default.recall("Default memory", top_k=5) |
| #297 | switch_results = mn_switch.recall("Switched memory", top_k=5) |
| #298 | |
| #299 | assert any("Default" in r["content"] for r in default_results) |
| #300 | assert any("Switched" in r["content"] for r in switch_results) |
| #301 | assert not any("Switched" in r["content"] for r in default_results) |
| #302 | assert not any("Default" in r["content"] for r in switch_results) |
| #303 | finally: |
| #304 | del os.environ["MNEMOSYNE_DATA_DIR"] |
| #305 | |
| #306 | def test_per_call_bank_parameter(self): |
| #307 | """Individual calls should accept bank parameter directly.""" |
| #308 | with tempfile.TemporaryDirectory() as tmpdir: |
| #309 | data_dir = Path(tmpdir) |
| #310 | os.environ["MNEMOSYNE_DATA_DIR"] = str(data_dir) |
| #311 | try: |
| #312 | mgr = BankManager(data_dir) |
| #313 | mgr.create_bank("call_test") |
| #314 | |
| #315 | import time |
| #316 | remember("Direct default " + str(time.time()), bank="default") |
| #317 | time.sleep(0.01) |
| #318 | remember("Direct custom " + str(time.time()), bank="call_test") |
| #319 | |
| #320 | default_results = recall("Direct default", bank="default") |
| #321 | custom_results = recall("Direct custom", bank="call_test") |
| #322 | |
| #323 | assert any("default" in r["content"] for r in default_results) |
| #324 | assert any("custom" in r["content"] for r in custom_results) |
| #325 | finally: |
| #326 | del os.environ["MNEMOSYNE_DATA_DIR"] |
| #327 | |
| #328 | def test_get_bank_returns_current(self): |
| #329 | """get_bank() should return the currently set bank.""" |
| #330 | original = get_bank() |
| #331 | set_bank("test_bank") |
| #332 | assert get_bank() == "test_bank" |
| #333 | set_bank(original) # Restore |
| #334 | |
| #335 | |
| #336 | # ============================================================================ |
| #337 | # Edge cases |
| #338 | # ============================================================================ |
| #339 | |
| #340 | class TestBankEdgeCases: |
| #341 | """Boundary conditions and error handling.""" |
| #342 | |
| #343 | def test_empty_bank_name_raises(self): |
| #344 | """Empty bank name should raise ValueError.""" |
| #345 | with tempfile.TemporaryDirectory() as tmpdir: |
| #346 | mgr = BankManager(Path(tmpdir)) |
| #347 | with pytest.raises(ValueError): |
| #348 | mgr.create_bank("") |
| #349 | |
| #350 | def test_long_bank_name_raises(self): |
| #351 | """Bank name > 64 chars should raise ValueError.""" |
| #352 | with tempfile.TemporaryDirectory() as tmpdir: |
| #353 | mgr = BankManager(Path(tmpdir)) |
| #354 | with pytest.raises(ValueError, match="exceeds 64"): |
| #355 | mgr.create_bank("a" * 65) |
| #356 | |
| #357 | def test_bank_name_with_hyphens_and_underscores(self): |
| #358 | """Hyphens and underscores should be valid.""" |
| #359 | with tempfile.TemporaryDirectory() as tmpdir: |
| #360 | mgr = BankManager(Path(tmpdir)) |
| #361 | mgr.create_bank("my-bank_1") |
| #362 | assert mgr.bank_exists("my-bank_1") |
| #363 | |
| #364 | def test_mnemosyne_db_path_override_takes_precedence(self): |
| #365 | """Explicit db_path should override bank resolution.""" |
| #366 | with tempfile.TemporaryDirectory() as tmpdir: |
| #367 | custom_db = Path(tmpdir) / "custom.db" |
| #368 | mn = Mnemosyne(db_path=custom_db, bank="work") |
| #369 | assert mn.db_path == custom_db |
| #370 | assert mn.bank == "work" # Bank is still tracked |
| #371 | |
| #372 | |
| #373 | # ============================================================================ |
| #374 | # Run standalone |
| #375 | # ============================================================================ |
| #376 | |
| #377 | if __name__ == "__main__": |
| #378 | pytest.main([__file__, "-v", "--tb=short"]) |
| #379 |