fix: resolve BW state sync issue -- add sync() to login lifecycle

The provisioner's BitwardenHelper.login() was missing the critical
`bw sync` step that the host wrapper includes. Without syncing after
login, the container's local vault cache was empty/stale, causing items
to vanish between container runs. Added sync() call at end of login()
and before list_items().

Also fixed container UID/GID to match host user (1002:1002) for proper
bind-mount access, and added source-code volume mounts for fast iteration.

Verified with 5-phase cross-container persistence test (create in
container A, verify in fresh container B, update in C, confirm in D).
This commit is contained in:
TSYS Group COO
2026-08-13 20:49:39 -05:00
parent c0eb1b383b
commit f633a10f80
4 changed files with 146 additions and 2 deletions
+127
View File
@@ -0,0 +1,127 @@
#!/usr/bin/env python3
"""
test_bw_persistence.py -- Cross-container BW state persistence test.
Verifies that BW items created in one container run are visible in a
subsequent container run (the "state sync" issue from session 1).
This test is designed to be invoked twice:
Run 1 (create): python3 test_bw_persistence.py create
Run 2 (verify): python3 test_bw_persistence.py verify
Run 3 (update): python3 test_bw_persistence.py update
Run 4 (confirm): python3 test_bw_persistence.py confirm
Run 5 (cleanup): python3 test_bw_persistence.py cleanup
Each run is a SEPARATE container invocation. If the bw-state bind mount
and sync logic are working, run 2 will see the item created in run 1,
and run 4 will see the update from run 3.
Usage:
docker compose run --rm --entrypoint python3 provision \
test_bw_persistence.py <phase>
"""
import json
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
from bw_helper import BitwardenHelper
TEST_ITEM = "TEST-PERSISTENCE-CROSSCONTAINER"
TEST_PASSWORD_ORIG = "OriginalPersistPassword123!"
TEST_PASSWORD_UPDATED = "UpdatedPersistPassword456@"
def make_helper() -> BitwardenHelper:
return BitwardenHelper(
client_id=os.environ["BW_CLIENTID"],
client_secret=os.environ["BW_CLIENTSECRET"],
password=os.environ["BW_PASSWORD"],
server_url=os.environ.get("BW_SERVER", ""),
)
def phase_create():
bw = make_helper()
bw.login()
existing = bw.get_item_id(TEST_ITEM)
if existing:
bw._run_bw(["delete", "item", existing])
item_id = bw.create_item(
name=TEST_ITEM,
username="persist-test@example.com",
password=TEST_PASSWORD_ORIG,
uris=["https://persist.example.com"],
collection_name="test",
)
assert item_id, "create should return an ID"
print(f"CREATE_OK: item_id={item_id}")
print(f"HOST_SYNC_CHECK: run 'bw list items' on host to verify visibility")
def phase_verify():
bw = make_helper()
bw.login()
assert bw.item_exists(TEST_ITEM), \
"FRESH CONTAINER CANNOT SEE ITEM CREATED BY PREVIOUS CONTAINER"
pw = bw.get_item_password(TEST_ITEM)
assert pw == TEST_PASSWORD_ORIG, \
f"Password mismatch: expected {TEST_PASSWORD_ORIG}, got {pw}"
print("VERIFY_OK: item visible in fresh container run")
def phase_update():
bw = make_helper()
bw.login()
original_id = bw.get_item_id(TEST_ITEM)
assert original_id, "item must exist before update"
updated_id = bw.update_item(TEST_ITEM, password=TEST_PASSWORD_UPDATED)
assert updated_id == original_id, "update must preserve item ID"
print(f"UPDATE_OK: item_id={updated_id}")
def phase_confirm():
bw = make_helper()
bw.login()
pw = bw.get_item_password(TEST_ITEM)
assert pw == TEST_PASSWORD_UPDATED, \
f"Update did not persist: expected {TEST_PASSWORD_UPDATED}, got {pw}"
print("CONFIRM_OK: update visible in fresh container run")
def phase_cleanup():
bw = make_helper()
bw.login()
item_id = bw.get_item_id(TEST_ITEM)
if item_id:
bw._run_bw(["delete", "item", item_id])
assert not bw.item_exists(TEST_ITEM), "cleanup failed"
print("CLEANUP_OK")
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: test_bw_persistence.py <create|verify|update|confirm|cleanup>")
sys.exit(1)
phase = sys.argv[1]
phases = {
"create": phase_create,
"verify": phase_verify,
"update": phase_update,
"confirm": phase_confirm,
"cleanup": phase_cleanup,
}
if phase not in phases:
print(f"Unknown phase: {phase}")
sys.exit(1)
phases[phase]()
print("SUCCESS")