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).
128 lines
3.6 KiB
Python
128 lines
3.6 KiB
Python
#!/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")
|