From 87e755c090a35871a8174348d223c449e2990d0e Mon Sep 17 00:00:00 2001 From: yuli Date: Thu, 20 Aug 2026 18:04:36 +0300 Subject: [PATCH 1/7] Omit unset options from request bodies, add per-field tokenization tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit json.dumps renders Python None as JSON null, so a key assigned options.get(k) was still present on the wire. The server builds its required-parameter set by testing key presence, so a null counted as supplied and got validated: - create_shared_record sent "appname": null and was rejected every time with "The appname parameter has an incorrect format", making the method unusable unless the caller passed an appname. - create_legal_basis sent "requiredflag": null, which panicked the server outright. The sibling SDKs avoid this for free — JSON.stringify drops undefined, PHP guards with isset(), Java checks != null — so Python needs the guard spelled out. _add_options() copies only supplied options into the body; an AST scan for keys assigned a possibly-None value found seven methods, all now routed through it. tests/test_field_tokenization.py covers the two flows a customer needs for per-field tokenization, each with single and bulk requests: - Per-field shared identities: a 10-field profile exposed through one SharedRecordCreate UUID per field, redeemed with no X-Bunker-Token, including field scoping and expiry. - Format-preserving tokenization: every value stringified through TokenCreate/TokenCreateBulk with unique set, round-tripped, with Luhn and length checks on the card token. Two tests track server-side work that is fixed in source but not yet deployed to pro.databunker.org: the bulk `unique` dedup assertion fails, and the `string` type alias skips. Co-Authored-By: Claude Opus 5 --- databunkerpro/api.py | 87 +++--- tests/test_field_tokenization.py | 518 +++++++++++++++++++++++++++++++ 2 files changed, 560 insertions(+), 45 deletions(-) create mode 100644 tests/test_field_tokenization.py diff --git a/databunkerpro/api.py b/databunkerpro/api.py index 6a659f3..ea2fbb8 100644 --- a/databunkerpro/api.py +++ b/databunkerpro/api.py @@ -3,7 +3,7 @@ import json import re from datetime import datetime -from typing import Any, Dict, List, Optional, TypedDict, Union, cast +from typing import Any, Dict, Iterable, List, Mapping, Optional, TypedDict, Union, cast import requests @@ -162,6 +162,24 @@ class PatchOperation(TypedDict, total=False): value: Optional[Any] # New value for the field +def _add_options( + data: Dict[str, Any], options: Optional[Mapping[str, Any]], keys: Iterable[str] +) -> Dict[str, Any]: + """Copy the supplied `keys` from `options` into `data`, skipping unset ones. + + An option the caller left out must stay out of the request body. Python's None + serializes to JSON null, and the server treats a null as a supplied value: it + validates `appname: null` and rejects it, and it panics on `requiredflag: null`. + Other language SDKs avoid this for free — JSON.stringify drops undefined — so + the guard has to be explicit here. + """ + for key in keys: + value = options.get(key) if options else None + if value is not None: + data[key] = value + return data + + class DatabunkerproAPI: """Main client class for interacting with the DatabunkerPro API.""" @@ -827,16 +845,20 @@ def create_legal_basis( request_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Create a legal basis for data processing.""" - data = { - "brief": options.get("brief"), - "status": options.get("status"), - "module": options.get("module"), - "fulldesc": options.get("fulldesc"), - "shortdesc": options.get("shortdesc"), - "basistype": options.get("basistype"), - "requiredmsg": options.get("requiredmsg"), - "requiredflag": options.get("requiredflag"), - } + data = _add_options( + {}, + options, + ( + "brief", + "status", + "module", + "fulldesc", + "shortdesc", + "basistype", + "requiredmsg", + "requiredflag", + ), + ) return self._make_request("LegalBasisCreate", data, request_metadata) def update_legal_basis( @@ -985,13 +1007,9 @@ def create_processing_activity( request_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Create a new processing activity.""" - data = { - "activity": options.get("activity"), - "title": options.get("title"), - "script": options.get("script"), - "fulldesc": options.get("fulldesc"), - "applicableto": options.get("applicableto"), - } + data = _add_options( + {}, options, ("activity", "title", "script", "fulldesc", "applicableto") + ) return self._make_request("ProcessingActivityCreate", data, request_metadata) def update_processing_activity( @@ -1058,11 +1076,7 @@ def create_group( request_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Create a new group.""" - data = { - "groupname": options.get("groupname"), - "groupdesc": options.get("groupdesc"), - "grouptype": options.get("grouptype"), - } + data = _add_options({}, options, ("groupname", "groupdesc", "grouptype")) return self._make_request("GroupCreate", data, request_metadata) def get_group( @@ -1226,11 +1240,7 @@ def create_tenant( self, options: TenantOptions, request_metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """Create a new tenant.""" - data = { - "tenantname": options.get("tenantname"), - "tenantorg": options.get("tenantorg"), - "email": options.get("email"), - } + data = _add_options({}, options, ("tenantname", "tenantorg", "email")) return self._make_request("TenantCreate", data, request_metadata) def get_tenant( @@ -1288,10 +1298,7 @@ def create_role( request_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Create a new role.""" - data = { - "rolename": options.get("rolename"), - "roledesc": options.get("roledesc"), - } + data = _add_options({}, options, ("rolename", "roledesc")) return self._make_request("RoleCreate", data, request_metadata) def update_role( @@ -1333,11 +1340,7 @@ def create_policy( request_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Create a new policy.""" - data = { - "policyname": options.get("policyname"), - "policydesc": options.get("policydesc"), - "policy": options.get("policy"), - } + data = _add_options({}, options, ("policyname", "policydesc", "policy")) return self._make_request("PolicyCreate", data, request_metadata) def update_policy( @@ -1729,14 +1732,8 @@ def create_shared_record( request_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Creates a shared record for a user.""" - data = { - "mode": mode, - "identity": identity, - "fields": options.get("fields") if options else None, - "partner": options.get("partner") if options else None, - "appname": options.get("appname") if options else None, - "finaltime": options.get("finaltime") if options else None, - } + data: Dict[str, Any] = {"mode": mode, "identity": identity} + _add_options(data, options, ("fields", "partner", "appname", "finaltime")) return self._make_request("SharedRecordCreate", data, request_metadata) def get_shared_record( diff --git a/tests/test_field_tokenization.py b/tests/test_field_tokenization.py new file mode 100644 index 0000000..d99571c --- /dev/null +++ b/tests/test_field_tokenization.py @@ -0,0 +1,518 @@ +""" +Per-field tokenization and per-field shared identities. + +Written for a customer who wants to tokenize *individual* fields rather than whole +profiles. Two flows, each exercised twice — once with single requests, once with the +bulk endpoints: + +1. Shared identities — a 10-field profile is stored once, then each field is exposed + through its own `SharedRecordCreate` UUID with its own expiration. The UUID is then + redeemed with **no** `X-Bunker-Token` at all, which is the point of the feature: the + partner holding the UUID never receives a vault credential. + +2. Format-preserving tokenization — every field value is stringified and pushed through + `TokenCreate` / `TokenCreateBulk` with `unique` set, so repeated values collapse onto + one token. + +Runs against https://pro.databunker.org with a throwaway sandbox tenant unless +DATABUNKER_API_URL / DATABUNKER_API_TOKEN / DATABUNKER_TENANT_NAME are set. +""" + +import os +import random +import time +import unittest + +import requests + +from databunkerpro import DatabunkerproAPI + +# The server accepts creditcard, email, text (alias: string), ssn, uuid, unixtimestamp, +# uint64 and uint32. Of those, only these four actually return a non-empty, format- +# preserving `tokenbase`; email, ssn and uuid come back with none despite not being +# plain strings. See the mapping in _token_type below. +FORMAT_PRESERVING_TYPES = {"creditcard", "unixtimestamp", "uint64", "uint32"} + + +def make_profile(seed: int) -> dict: + """A 10-field profile — the customer's record shape.""" + return { + "email": f"jane{seed}@example.com", + "phone": f"+1206555{seed % 10000:04d}", + "login": f"jane{seed}", + "custom": f"cust-{seed}", + "first": "Jane", + "last": f"Doe{seed}", + "address": "1 Market Street, San Francisco, CA", + "dob": "1985-04-12", + "creditcard": "4532015112830366", + "ssn": f"123-45-{seed % 10000:04d}", + } + + +def is_luhn_valid(number: str) -> bool: + """Luhn checksum — a format-preserving card token must still pass it.""" + digits = [int(c) for c in number][::-1] + total = 0 + for i, digit in enumerate(digits): + if i % 2: + digit *= 2 + if digit > 9: + digit -= 9 + total += digit + return total % 10 == 0 + + +class SandboxTestCase(unittest.TestCase): + """Shared sandbox bootstrap, matching the other test modules.""" + + @classmethod + def setUpClass(cls): + cls.api_url = os.getenv("DATABUNKER_API_URL", "https://pro.databunker.org") + cls.api_token = os.getenv("DATABUNKER_API_TOKEN", "") + cls.tenant_name = os.getenv("DATABUNKER_TENANT_NAME", "") + + if not all([cls.api_token, cls.tenant_name]): + try: + response = requests.get( + "https://databunker.org/api/newtenant.php", verify=False + ) + data = response.json() if response.ok else None + if not data or data.get("status") != "ok": + raise unittest.SkipTest("Failed to get credentials from sandbox") + cls.tenant_name = data["tenantname"] + cls.api_token = data["xtoken"] + print(f"\nSandbox tenant: {cls.tenant_name} at {cls.api_url}") + except unittest.SkipTest: + raise + except Exception as e: + raise unittest.SkipTest(f"Failed to get credentials: {e}") + + cls.api = DatabunkerproAPI(cls.api_url, cls.api_token, cls.tenant_name) + # A client with no X-Bunker-Token. The tenant header stays, because without it + # the server cannot route the request and answers 404 for every UUID — that is + # addressing, not authentication. + cls.anonymous = DatabunkerproAPI(cls.api_url, "", cls.tenant_name) + cls.seed = random.randint(100000, 999999) + + +class TestSharedIdentityPerField(SandboxTestCase): + """One shared-record UUID per profile field, redeemed without a credential.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.profile = make_profile(cls.seed) + + created = cls.api.create_user(cls.profile) + if created.get("status") != "ok": + raise unittest.SkipTest(f"Setup UserCreate failed: {created}") + cls.user_token = created["token"] + + # The user access token the customer hands to the end user's own session. + cls.xtoken_result = cls.api.create_user_x_token( + "token", cls.user_token, {"finaltime": "10m"} + ) + + # One shared identity per field, each with its own expiration and partner tag. + cls.shares = {} + for field in cls.profile: + result = cls.api.create_shared_record( + "token", + cls.user_token, + { + "fields": field, + "partner": f"partner-{field}", + "finaltime": "1d", + }, + ) + cls.shares[field] = result + + def test_profile_has_ten_fields(self): + self.assertEqual(len(self.profile), 10) + + def test_user_x_token_reads_own_record(self): + """XTokenCreateForUser issues a working, scoped credential.""" + self.assertEqual( + self.xtoken_result.get("status"), "ok", f"{self.xtoken_result}" + ) + self.assertIn("xtoken", self.xtoken_result) + + user_api = DatabunkerproAPI( + self.api_url, self.xtoken_result["xtoken"], self.tenant_name + ) + result = user_api.get_user("token", self.user_token) + self.assertEqual(result.get("status"), "ok", f"{result}") + self.assertEqual(result.get("profile"), self.profile) + + def test_every_field_gets_its_own_share(self): + """All 10 SharedRecordCreate calls succeed and return distinct UUIDs.""" + for field, result in self.shares.items(): + with self.subTest(field=field): + self.assertEqual( + result.get("status"), "ok", f"share for {field}: {result}" + ) + self.assertIn("recorduuid", result) + + uuids = [r["recorduuid"] for r in self.shares.values() if "recorduuid" in r] + self.assertEqual(len(uuids), 10) + self.assertEqual(len(set(uuids)), 10, "shared record UUIDs must be unique") + + def test_shared_identity_resolves_without_authentication(self): + """The whole point: SharedRecordGet needs no X-Bunker-Token.""" + for field, share in self.shares.items(): + with self.subTest(field=field): + self.assertIn("recorduuid", share, f"share for {field}: {share}") + result = self.anonymous.get_shared_record(share["recorduuid"]) + self.assertEqual( + result.get("status"), + "ok", + f"unauthenticated read of {field} failed: {result}", + ) + self.assertEqual(result.get("data"), {field: self.profile[field]}) + + def test_share_exposes_only_its_own_field(self): + """Field scoping is enforced — no other field leaks into the response.""" + for field, share in self.shares.items(): + with self.subTest(field=field): + data = self.anonymous.get_shared_record(share["recorduuid"]).get( + "data", {} + ) + self.assertEqual(list(data), [field], f"{field} share leaked: {data}") + + def test_unauthenticated_client_cannot_read_the_user_record(self): + """The shared UUID is the only door — plain UserGet still requires a token.""" + result = self.anonymous.get_user("token", self.user_token) + self.assertNotEqual( + result.get("status"), + "ok", + f"unauthenticated UserGet returned data: {result}", + ) + + def test_unknown_uuid_is_rejected(self): + result = self.anonymous.get_shared_record( + "00000000-0000-0000-0000-000000000000" + ) + self.assertNotEqual(result.get("status"), "ok", f"{result}") + + def test_share_stops_resolving_after_finaltime(self): + """A short-lived share expires on its own, with no clean-up call.""" + share = self.api.create_shared_record( + "token", + self.user_token, + {"fields": "first", "partner": "partner-expiry", "finaltime": "2s"}, + ) + self.assertEqual(share.get("status"), "ok", f"{share}") + + before = self.anonymous.get_shared_record(share["recorduuid"]) + self.assertEqual( + before.get("status"), "ok", f"share was born expired: {before}" + ) + + time.sleep(4) + + after = self.anonymous.get_shared_record(share["recorduuid"]) + self.assertNotEqual( + after.get("status"), "ok", f"expired share still resolves: {after}" + ) + + +class TestSharedIdentityPerFieldBulk(SandboxTestCase): + """Same flow, with the users created through UserCreateBulk.""" + + USER_COUNT = 3 + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.profiles = [make_profile(cls.seed + i) for i in range(cls.USER_COUNT)] + + result = cls.api.create_users_bulk( + [{"profile": p} for p in cls.profiles], {"finaltime": "1d"} + ) + if result.get("status") != "ok": + raise unittest.SkipTest(f"Setup UserCreateBulk failed: {result}") + cls.bulk_result = result + + # email -> user token, so a share can be matched back to its source profile. + cls.tokens = { + item["profile"]["email"]: item["token"] for item in result["created"] + } + + # (email, field) -> recorduuid + cls.shares = {} + for profile in cls.profiles: + token = cls.tokens[profile["email"]] + for field in profile: + share = cls.api.create_shared_record( + "token", + token, + { + "fields": field, + "partner": f"partner-{field}", + "finaltime": "1d", + }, + ) + cls.shares[(profile["email"], field)] = share + + def test_bulk_create_returns_a_token_per_user(self): + self.assertEqual(len(self.bulk_result["created"]), self.USER_COUNT) + self.assertEqual(len(set(self.tokens.values())), self.USER_COUNT) + + def test_every_field_of_every_user_is_shared(self): + self.assertEqual(len(self.shares), self.USER_COUNT * 10) + uuids = [s["recorduuid"] for s in self.shares.values() if "recorduuid" in s] + self.assertEqual(len(uuids), self.USER_COUNT * 10, "some shares failed") + self.assertEqual(len(set(uuids)), len(uuids), "UUIDs collided across users") + + def test_each_share_resolves_unauthenticated_to_its_own_value(self): + by_email = {p["email"]: p for p in self.profiles} + for (email, field), share in self.shares.items(): + with self.subTest(email=email, field=field): + self.assertIn("recorduuid", share, f"{share}") + result = self.anonymous.get_shared_record(share["recorduuid"]) + self.assertEqual(result.get("status"), "ok", f"{result}") + self.assertEqual(result.get("data"), {field: by_email[email][field]}) + + +class TestFieldTokenization(SandboxTestCase): + """TokenCreate per field, one request at a time, with `unique` set.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.profile = make_profile(cls.seed) + cls.tokens = { + field: cls.api.create_token( + cls._token_type(field), str(value), {"unique": True} + ) + for field, value in cls.profile.items() + } + + @staticmethod + def _token_type(field: str) -> str: + """Every value is stringified and tokenized as a string. + + `text` is the API's string type, and `string` is an accepted alias for it. + The test sends the canonical `text`, which every server version understands; + test_string_is_an_alias_for_text covers the alias itself. Email, dates and + timestamps all go through it rather than a specialised type. The credit card + is the exception: it is the one field where format preservation applies. + """ + return "creditcard" if field == "creditcard" else "text" + + def test_every_field_is_tokenized(self): + for field, result in self.tokens.items(): + with self.subTest(field=field): + self.assertEqual(result.get("status"), "ok", f"{field}: {result}") + self.assertTrue(result.get("tokenuuid"), f"{field}: {result}") + + def test_tokens_are_distinct_across_fields(self): + uuids = [r["tokenuuid"] for r in self.tokens.values() if r.get("tokenuuid")] + self.assertEqual(len(uuids), 10) + self.assertEqual(len(set(uuids)), 10, "two fields collapsed onto one token") + + def test_detokenization_round_trips_every_field(self): + for field, value in self.profile.items(): + with self.subTest(field=field): + result = self.api.get_token(self.tokens[field]["tokenuuid"]) + self.assertEqual(result.get("status"), "ok", f"{field}: {result}") + self.assertEqual(result.get("record"), str(value)) + self.assertEqual(result.get("tokentype"), self._token_type(field)) + + def test_credit_card_token_preserves_format(self): + """`tokenbase` is same-length, all-digits and Luhn-valid — but not the card.""" + original = self.profile["creditcard"] + tokenbase = self.tokens["creditcard"].get("tokenbase") + + self.assertTrue(tokenbase, f"no tokenbase: {self.tokens['creditcard']}") + self.assertNotEqual(tokenbase, original, "token equals the original card") + self.assertEqual(len(tokenbase), len(original)) + self.assertTrue(tokenbase.isdigit()) + self.assertTrue(is_luhn_valid(tokenbase), f"{tokenbase} fails the Luhn check") + + def test_format_preserving_token_also_detokenizes(self): + """`tokenbase` works as a lookup key, not only `tokenuuid`.""" + result = self.api.get_token(self.tokens["creditcard"]["tokenbase"]) + self.assertEqual(result.get("status"), "ok", f"{result}") + self.assertEqual(result.get("record"), self.profile["creditcard"]) + + def test_only_creditcard_returns_a_format_preserving_token(self): + """Documents current coverage: email/text get a UUID and an empty tokenbase.""" + for field, result in self.tokens.items(): + with self.subTest(field=field): + if self._token_type(field) in FORMAT_PRESERVING_TYPES: + self.assertTrue(result.get("tokenbase"), f"{field}: {result}") + else: + self.assertFalse(result.get("tokenbase"), f"{field}: {result}") + + def test_unique_returns_the_same_token_for_the_same_value(self): + """`unique` is the dedup flag — re-tokenizing a field is idempotent.""" + for field, value in self.profile.items(): + with self.subTest(field=field): + again = self.api.create_token( + self._token_type(field), str(value), {"unique": True} + ) + self.assertEqual(again.get("status"), "ok", f"{field}: {again}") + self.assertEqual( + again["tokenuuid"], + self.tokens[field]["tokenuuid"], + f"{field}: unique=True issued a second token for one value", + ) + + def test_without_unique_each_call_mints_a_new_token(self): + first = self.api.create_token("creditcard", self.profile["creditcard"]) + second = self.api.create_token("creditcard", self.profile["creditcard"]) + self.assertEqual(first.get("status"), "ok", f"{first}") + self.assertEqual(second.get("status"), "ok", f"{second}") + self.assertNotEqual(first["tokenuuid"], second["tokenuuid"]) + + def test_string_is_an_alias_for_text(self): + """ "string" and "text" are one type — same canonical name, same token. + + Skips on servers predating the alias, so the file works against a + pro.databunker.org that has not picked up the change yet. + """ + value = f"alias-probe-{self.seed}" + canonical = self.api.create_token("text", value, {"unique": True}) + self.assertEqual(canonical.get("status"), "ok", f"{canonical}") + + alias = self.api.create_token("string", value, {"unique": True}) + if alias.get("message") == "Wrong token type": + self.skipTest("server does not support the `string` alias yet") + + self.assertEqual(alias.get("status"), "ok", f"{alias}") + self.assertEqual( + alias["tokenuuid"], + canonical["tokenuuid"], + "`string` did not resolve to the same token as `text`", + ) + stored = self.api.get_token(alias["tokenuuid"]) + self.assertEqual(stored.get("tokentype"), "text", f"{stored}") + + def test_unknown_token_types_are_rejected(self): + """The accepted set is closed — these are not token types.""" + for name in ("phone", "date", "unixtime", "str"): + with self.subTest(tokentype=name): + result = self.api.create_token(name, f"probe-{self.seed}") + self.assertNotEqual(result.get("status"), "ok", f"{name}: {result}") + + +class TestFieldTokenizationBulk(SandboxTestCase): + """The same 10 fields through TokenCreateBulk, then bulk detokenization.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.profile = make_profile(cls.seed) + cls.records = [ + {"tokentype": TestFieldTokenization._token_type(f), "record": str(v)} + for f, v in cls.profile.items() + ] + cls.bulk = cls.api.create_tokens_bulk(cls.records, {"unique": True}) + + def _rows(self, result): + """created + duplicates — both carry a usable token for the record.""" + return (result.get("created") or []) + (result.get("duplicates") or []) + + def test_bulk_accepts_every_field(self): + self.assertEqual(self.bulk.get("status"), "ok", f"{self.bulk}") + self.assertEqual( + self.bulk.get("summary", {}).get("errors", 0), + 0, + f"TokenCreateBulk rejected records: {self.bulk.get('errors')}", + ) + self.assertEqual(len(self._rows(self.bulk)), len(self.records)) + + def test_bulk_round_trips_through_bulk_list_tokens(self): + """BulkListTokens returns the original value for every token in the batch.""" + rows = self._rows(self.bulk) + if not rows: + self.skipTest("no tokens were created") + + unlock = self.api.bulk_list_unlock() + self.assertEqual(unlock.get("status"), "ok", f"unlock failed: {unlock}") + + uuids = [row["tokenuuid"] for row in rows] + result = self.api.bulk_list_tokens(unlock["unlockuuid"], uuids) + self.assertEqual(result.get("status"), "ok", f"{result}") + + detokenized = { + row["tokenuuid"]: row["record"] for row in result.get("rows", []) + } + self.assertEqual( + len(detokenized), len(uuids), f"BulkListTokens returned {len(detokenized)}" + ) + for row in rows: + self.assertEqual(detokenized.get(row["tokenuuid"]), row["record"]) + + def test_bulk_credit_card_token_preserves_format(self): + card = self.profile["creditcard"] + row = next((r for r in self._rows(self.bulk) if r["record"] == card), None) + self.assertIsNotNone(row, f"credit card missing from batch: {self.bulk}") + + tokenbase = row.get("tokenbase") + self.assertTrue(tokenbase, f"no tokenbase: {row}") + self.assertNotEqual(tokenbase, card) + self.assertEqual(len(tokenbase), len(card)) + self.assertTrue(is_luhn_valid(tokenbase), f"{tokenbase} fails the Luhn check") + + def test_bulk_unique_deduplicates_a_repeated_batch(self): + """Re-submitting the batch with `unique` must reuse the existing tokens.""" + again = self.api.create_tokens_bulk(self.records, {"unique": True}) + self.assertEqual(again.get("status"), "ok", f"{again}") + + first = {r["record"]: r["tokenuuid"] for r in self._rows(self.bulk)} + second = {r["record"]: r["tokenuuid"] for r in self._rows(again)} + + self.assertEqual( + second, + first, + "unique=True minted new tokens for values that were already tokenized", + ) + + def test_single_create_finds_tokens_that_bulk_created(self): + """The dedup *lookup* works for `text` — so the batch write path is at fault. + + Pairs with test_bulk_unique_deduplicates_a_repeated_batch: a single + TokenCreate with unique=True returns the very token TokenCreateBulk stored, + which means the value is indexed and findable. Only the bulk writer skips + the uniqueness check. + """ + rows = self._rows(self.bulk) + if not rows: + self.skipTest("no tokens were created") + + for row in rows: + with self.subTest(record=row["record"]): + single = self.api.create_token( + row["tokentype"], row["record"], {"unique": True} + ) + self.assertEqual(single.get("status"), "ok", f"{single}") + self.assertEqual( + single["tokenuuid"], + row["tokenuuid"], + "single TokenCreate did not find the bulk-created token", + ) + + def test_bulk_delete_removes_the_tokens(self): + batch = [ + {"tokentype": "text", "record": f"disposable-{self.seed}-{i}"} + for i in range(3) + ] + created = self.api.create_tokens_bulk(batch) + self.assertEqual(created.get("status"), "ok", f"{created}") + uuids = [row["tokenuuid"] for row in created.get("created") or []] + self.assertEqual(len(uuids), 3, f"{created}") + + unlock = self.api.bulk_list_unlock() + self.assertEqual(unlock.get("status"), "ok", f"{unlock}") + deleted = self.api.bulk_delete_tokens(unlock["unlockuuid"], uuids) + self.assertEqual(deleted.get("status"), "ok", f"{deleted}") + + gone = self.api.get_token(uuids[0]) + self.assertNotEqual(gone.get("status"), "ok", f"token survived delete: {gone}") + + +if __name__ == "__main__": + unittest.main() From 10d331cd584938b386fb9c85c8f47a8faf04590f Mon Sep 17 00:00:00 2001 From: yuli Date: Thu, 20 Aug 2026 18:05:14 +0300 Subject: [PATCH 2/7] improve gitignore --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index 711e1b5..985f5aa 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ __pycache__ +dist databunkerpro.egg-info From 608fdf2cd68b0bb5eb90f392ae41eb8d09be9e13 Mon Sep 17 00:00:00 2001 From: yuli Date: Thu, 20 Aug 2026 18:05:56 +0300 Subject: [PATCH 3/7] Bump version to 0.1.7 Co-Authored-By: Claude Opus 5 --- databunkerpro/__init__.py | 2 +- setup.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/databunkerpro/__init__.py b/databunkerpro/__init__.py index 7e91b9a..3b8614d 100644 --- a/databunkerpro/__init__.py +++ b/databunkerpro/__init__.py @@ -5,5 +5,5 @@ from .api import DatabunkerproAPI -__version__ = "0.1.6" +__version__ = "0.1.7" __all__ = ["DatabunkerproAPI"] diff --git a/setup.py b/setup.py index 5ecd55f..028d9b2 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="databunkerpro", - version="0.1.6", + version="0.1.7", author="Databunker team", author_email="hello@databunker.org", description="Python client library for DatabunkerPro API", From 8e87ea93fb2b9fc8353314780de75ee9e0587a31 Mon Sep 17 00:00:00 2001 From: yuli Date: Thu, 20 Aug 2026 18:09:15 +0300 Subject: [PATCH 4/7] Narrow the format-preservation check to the card field FORMAT_PRESERVING_TYPES listed unixtimestamp, uint64 and uint32 alongside creditcard, but _token_type only ever returns creditcard or text, so those three could never be reached. The set documented the server's wider capabilities rather than anything this file exercises. Drop it and test the one distinction that applies here: the card gets a tokenbase, every string field does not. Co-Authored-By: Claude Opus 5 --- tests/test_field_tokenization.py | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/tests/test_field_tokenization.py b/tests/test_field_tokenization.py index d99571c..535bce6 100644 --- a/tests/test_field_tokenization.py +++ b/tests/test_field_tokenization.py @@ -27,12 +27,6 @@ from databunkerpro import DatabunkerproAPI -# The server accepts creditcard, email, text (alias: string), ssn, uuid, unixtimestamp, -# uint64 and uint32. Of those, only these four actually return a non-empty, format- -# preserving `tokenbase`; email, ssn and uuid come back with none despite not being -# plain strings. See the mapping in _token_type below. -FORMAT_PRESERVING_TYPES = {"creditcard", "unixtimestamp", "uint64", "uint32"} - def make_profile(seed: int) -> dict: """A 10-field profile — the customer's record shape.""" @@ -337,11 +331,11 @@ def test_format_preserving_token_also_detokenizes(self): self.assertEqual(result.get("status"), "ok", f"{result}") self.assertEqual(result.get("record"), self.profile["creditcard"]) - def test_only_creditcard_returns_a_format_preserving_token(self): - """Documents current coverage: email/text get a UUID and an empty tokenbase.""" + def test_only_the_card_returns_a_format_preserving_token(self): + """The card gets a `tokenbase`; every string field gets a UUID token only.""" for field, result in self.tokens.items(): with self.subTest(field=field): - if self._token_type(field) in FORMAT_PRESERVING_TYPES: + if field == "creditcard": self.assertTrue(result.get("tokenbase"), f"{field}: {result}") else: self.assertFalse(result.get("tokenbase"), f"{field}: {result}") From 31d2b74b801759a5dbbc12ce167d442943f34e2d Mon Sep 17 00:00:00 2001 From: yuli Date: Thu, 20 Aug 2026 18:14:09 +0300 Subject: [PATCH 5/7] Drop XTokenCreateForUser from the shared identity test The customer's "create a user token" is the record token UserCreate returns, which is what the shared records are keyed on. Minting a login xtoken as well tested authenticated access, which is beside the point of a flow whose whole premise is retrieval without a credential. Replace it with a check that the record token reads back the stored profile. Co-Authored-By: Claude Opus 5 --- tests/test_field_tokenization.py | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/tests/test_field_tokenization.py b/tests/test_field_tokenization.py index 535bce6..8b01582 100644 --- a/tests/test_field_tokenization.py +++ b/tests/test_field_tokenization.py @@ -103,11 +103,6 @@ def setUpClass(cls): raise unittest.SkipTest(f"Setup UserCreate failed: {created}") cls.user_token = created["token"] - # The user access token the customer hands to the end user's own session. - cls.xtoken_result = cls.api.create_user_x_token( - "token", cls.user_token, {"finaltime": "10m"} - ) - # One shared identity per field, each with its own expiration and partner tag. cls.shares = {} for field in cls.profile: @@ -125,17 +120,9 @@ def setUpClass(cls): def test_profile_has_ten_fields(self): self.assertEqual(len(self.profile), 10) - def test_user_x_token_reads_own_record(self): - """XTokenCreateForUser issues a working, scoped credential.""" - self.assertEqual( - self.xtoken_result.get("status"), "ok", f"{self.xtoken_result}" - ) - self.assertIn("xtoken", self.xtoken_result) - - user_api = DatabunkerproAPI( - self.api_url, self.xtoken_result["xtoken"], self.tenant_name - ) - result = user_api.get_user("token", self.user_token) + def test_user_token_reads_the_stored_profile(self): + """UserCreate returns the token the shared records are then keyed on.""" + result = self.api.get_user("token", self.user_token) self.assertEqual(result.get("status"), "ok", f"{result}") self.assertEqual(result.get("profile"), self.profile) From 9c411933570e262ef5718ec8f515df4205215561 Mon Sep 17 00:00:00 2001 From: yuli Date: Thu, 20 Aug 2026 18:22:41 +0300 Subject: [PATCH 6/7] Stop exercising the "string" token type from the SDK test The test sent "string" only to prove it aliases "text", which made the file depend on a pro.databunker.org deploy and put a skipTest in the way of a real assertion. The alias belongs to the server and is covered there by TestV2StringTokenTypeAlias. Every field here goes through the canonical "text". Co-Authored-By: Claude Opus 5 --- tests/test_field_tokenization.py | 31 +++---------------------------- 1 file changed, 3 insertions(+), 28 deletions(-) diff --git a/tests/test_field_tokenization.py b/tests/test_field_tokenization.py index 8b01582..79a3b21 100644 --- a/tests/test_field_tokenization.py +++ b/tests/test_field_tokenization.py @@ -274,11 +274,9 @@ def setUpClass(cls): def _token_type(field: str) -> str: """Every value is stringified and tokenized as a string. - `text` is the API's string type, and `string` is an accepted alias for it. - The test sends the canonical `text`, which every server version understands; - test_string_is_an_alias_for_text covers the alias itself. Email, dates and - timestamps all go through it rather than a specialised type. The credit card - is the exception: it is the one field where format preservation applies. + `text` is the API's string type. Email, dates and timestamps all go through + it rather than a specialised type. The credit card is the exception: it is + the one field where format preservation applies. """ return "creditcard" if field == "creditcard" else "text" @@ -348,29 +346,6 @@ def test_without_unique_each_call_mints_a_new_token(self): self.assertEqual(second.get("status"), "ok", f"{second}") self.assertNotEqual(first["tokenuuid"], second["tokenuuid"]) - def test_string_is_an_alias_for_text(self): - """ "string" and "text" are one type — same canonical name, same token. - - Skips on servers predating the alias, so the file works against a - pro.databunker.org that has not picked up the change yet. - """ - value = f"alias-probe-{self.seed}" - canonical = self.api.create_token("text", value, {"unique": True}) - self.assertEqual(canonical.get("status"), "ok", f"{canonical}") - - alias = self.api.create_token("string", value, {"unique": True}) - if alias.get("message") == "Wrong token type": - self.skipTest("server does not support the `string` alias yet") - - self.assertEqual(alias.get("status"), "ok", f"{alias}") - self.assertEqual( - alias["tokenuuid"], - canonical["tokenuuid"], - "`string` did not resolve to the same token as `text`", - ) - stored = self.api.get_token(alias["tokenuuid"]) - self.assertEqual(stored.get("tokentype"), "text", f"{stored}") - def test_unknown_token_types_are_rejected(self): """The accepted set is closed — these are not token types.""" for name in ("phone", "date", "unixtime", "str"): From fc17846294c74943f14ee80b67d5cec191650f34 Mon Sep 17 00:00:00 2001 From: yuli Date: Thu, 20 Aug 2026 18:25:34 +0300 Subject: [PATCH 7/7] Revert "Bump version to 0.1.7" This reverts commit 608fdf2cd68b0bb5eb90f392ae41eb8d09be9e13. --- databunkerpro/__init__.py | 2 +- setup.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/databunkerpro/__init__.py b/databunkerpro/__init__.py index 3b8614d..7e91b9a 100644 --- a/databunkerpro/__init__.py +++ b/databunkerpro/__init__.py @@ -5,5 +5,5 @@ from .api import DatabunkerproAPI -__version__ = "0.1.7" +__version__ = "0.1.6" __all__ = ["DatabunkerproAPI"] diff --git a/setup.py b/setup.py index 028d9b2..5ecd55f 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="databunkerpro", - version="0.1.7", + version="0.1.6", author="Databunker team", author_email="hello@databunker.org", description="Python client library for DatabunkerPro API",