import io import urllib.error import unittest from contextlib import redirect_stdout from unittest.mock import patch from setup_auth_family import ( build_schema, ensure_collection, get_collection, merge_fields, merge_indexes, relation_field, ) class AuthFamilySchemaTests(unittest.TestCase): def test_schema_contract_covers_users_students_collections_and_rules(self): schema = build_schema() self.assertEqual(schema["users_fields"]["rol"]["name"], "rol") self.assertEqual( schema["users_fields"]["rol"]["values"], ["teacher", "student", "guardian"], ) self.assertIn("phone", schema["users_fields"]) self.assertIn("otp_enabled", schema["users_fields"]) self.assertIn("is_minor", schema["student_fields"]) self.assertNotIn("replacement_phone", schema["student_fields"]) self.assertEqual( set(schema["collections"]), {"guardian_profiles", "student_guardians", "otp_codes", "otp_identity_grants"}, ) self.assertEqual( schema["collections"]["student_guardians"]["indexes"], [ "CREATE UNIQUE INDEX idx_student_guardian ON student_guardians (student_id, guardian_id)", "CREATE INDEX idx_student_guardians_student ON student_guardians (student_id)", "CREATE INDEX idx_student_guardians_guardian ON student_guardians (guardian_id)", ], ) self.assertIn("eligible", { field["name"] for field in schema["collections"]["otp_codes"]["fields"] }) self.assertEqual( schema["collections"]["otp_codes"]["indexes"], [ "CREATE INDEX idx_otp_codes_phone_requested ON otp_codes (phone, requested_at)", "CREATE INDEX idx_otp_codes_ip_requested ON otp_codes (request_ip, requested_at)", ], ) self.assertEqual( schema["collections"]["guardian_profiles"]["indexes"], [ "CREATE UNIQUE INDEX idx_guardian_profiles_phone ON guardian_profiles (phone) WHERE phone != ''", "CREATE UNIQUE INDEX idx_guardian_profiles_user ON guardian_profiles (user_id)", ], ) self.assertEqual( schema["existing_collections"]["users"]["indexes"], [ "CREATE INDEX idx_users_phone ON users (phone)", "CREATE UNIQUE INDEX idx_users_guardian_phone ON users (phone) WHERE rol = 'guardian' AND phone != ''", ], ) self.assertEqual( schema["existing_collections"]["student_profiles"]["indexes"], [ "CREATE INDEX idx_student_profiles_phone ON student_profiles (phone)", "CREATE UNIQUE INDEX idx_student_profiles_user ON student_profiles (user_id)", ], ) self.assertIsNone(schema["collections"]["otp_codes"]["listRule"]) self.assertIsNone(schema["collections"]["otp_codes"]["viewRule"]) self.assertIsNone(schema["collections"]["otp_codes"]["createRule"]) self.assertIsNone(schema["collections"]["otp_codes"]["updateRule"]) self.assertIsNone(schema["collections"]["otp_codes"]["deleteRule"]) self.assertIsNone(schema["collections"]["otp_identity_grants"]["listRule"]) self.assertIsNone(schema["collections"]["otp_identity_grants"]["viewRule"]) self.assertIsNone(schema["collections"]["otp_identity_grants"]["createRule"]) self.assertIsNone(schema["collections"]["otp_identity_grants"]["updateRule"]) self.assertIsNone(schema["collections"]["otp_identity_grants"]["deleteRule"]) self.assertEqual( schema["existing_collections"]["users"]["listRule"], "@request.auth.id != '' && (@request.auth.rol = 'teacher' || id = @request.auth.id)", ) self.assertEqual( schema["existing_collections"]["users"]["viewRule"], "@request.auth.id != '' && (@request.auth.rol = 'teacher' || id = @request.auth.id)", ) self.assertEqual( schema["existing_collections"]["users"]["createRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["users"]["updateRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["users"]["deleteRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["student_profiles"]["listRule"], "@request.auth.id != '' && (@request.auth.rol = 'teacher' || user_id = @request.auth.id)", ) self.assertEqual( schema["existing_collections"]["student_profiles"]["viewRule"], "@request.auth.id != '' && (@request.auth.rol = 'teacher' || user_id = @request.auth.id)", ) self.assertEqual( schema["existing_collections"]["student_profiles"]["createRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["student_profiles"]["updateRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["student_profiles"]["deleteRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["classes"]["listRule"], "@request.auth.id != '' && (@request.auth.rol = 'teacher' || student_id = @request.auth.id)", ) self.assertEqual( schema["existing_collections"]["classes"]["viewRule"], "@request.auth.id != '' && (@request.auth.rol = 'teacher' || student_id = @request.auth.id)", ) self.assertEqual( schema["existing_collections"]["classes"]["createRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["classes"]["updateRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) self.assertEqual( schema["existing_collections"]["classes"]["deleteRule"], "@request.auth.id != '' && @request.auth.rol = 'teacher'", ) def test_relation_field_uses_expected_relation_shape(self): field = relation_field("guardian_id", "_pb_users_auth_", required=True, cascade=True) self.assertEqual( field, { "name": "guardian_id", "type": "relation", "required": True, "collectionId": "_pb_users_auth_", "maxSelect": 1, "cascadeDelete": True, }, ) def test_merge_fields_preserves_existing_fields_and_extends_select_values(self): current = [ {"name": "rol", "type": "select", "required": False, "maxSelect": 1, "values": ["teacher", "student"]}, {"name": "display_name", "type": "text", "required": False}, ] desired = build_schema()["users_fields"] merged, changed = merge_fields(current, desired) self.assertTrue(changed) self.assertEqual([field["name"] for field in merged], ["rol", "display_name", "phone", "otp_enabled"]) self.assertEqual(merged[0]["values"], ["teacher", "student", "guardian"]) self.assertEqual(merged[1]["name"], "display_name") merged_again, changed_again = merge_fields(merged, desired) self.assertFalse(changed_again) self.assertEqual(merged_again, merged) def test_merge_indexes_preserves_existing_indexes_and_is_idempotent(self): required = "CREATE UNIQUE INDEX idx_student_guardian ON student_guardians (student_id, guardian_id)" existing = [ "CREATE INDEX idx_existing ON student_guardians (student_id)", ] merged, changed = merge_indexes(existing, [required]) self.assertTrue(changed) self.assertEqual( merged, [ "CREATE INDEX idx_existing ON student_guardians (student_id)", required, ], ) merged_again, changed_again = merge_indexes(merged, [required]) self.assertFalse(changed_again) self.assertEqual(merged_again, merged) def test_merge_indexes_normalizes_spacing_and_case_variants(self): existing = [ "create unique INDEX idx_student_guardian on student_guardians ( student_id , guardian_id )" ] desired = [ "CREATE UNIQUE INDEX idx_student_guardian ON student_guardians (student_id, guardian_id)" ] merged, changed = merge_indexes(existing, desired) self.assertFalse(changed) self.assertEqual(merged, existing) def test_get_collection_returns_status_and_body(self): class FakeResponse: def __init__(self, status, payload): self.status = status self._payload = payload def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def read(self): return self._payload.encode("utf-8") def fake_client(_request): return FakeResponse(403, "{\"error\":\"forbidden\"}") data, status = get_collection("users", "token", request_client=fake_client) self.assertEqual(status, 403) self.assertEqual(data, {"error": "forbidden"}) def test_ensure_collection_creates_only_on_404(self): calls = [] class FakeResponse: def __init__(self, status, payload="{}"): self.status = status self._payload = payload def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def read(self): return self._payload.encode("utf-8") def fake_client(request): calls.append((request.get_method(), request.full_url)) if request.get_method() == "GET": return FakeResponse(404, "{\"error\":\"not found\"}") if request.get_method() == "POST": return FakeResponse(201, "{\"id\":\"guardian_profiles\"}") raise AssertionError(f"unexpected method {request.get_method()}") with redirect_stdout(io.StringIO()) as buffer: ensure_collection( "guardian_profiles", {"type": "base", "fields": []}, "token", create_if_missing=True, request_client=fake_client, ) self.assertEqual(buffer.getvalue().strip(), "guardian_profiles") self.assertEqual([method for method, _ in calls], ["GET", "POST"]) def test_ensure_collection_rejects_403_without_create(self): calls = [] class FakeResponse: def __init__(self, status, payload): self.status = status self._payload = payload def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def read(self): return self._payload.encode("utf-8") def fake_client(request): calls.append(request.get_method()) if request.get_method() == "GET": return FakeResponse(403, "{\"error\":\"forbidden\"}") if request.get_method() == "POST": raise AssertionError("create must not be attempted on 403") raise AssertionError(f"unexpected method {request.get_method()}") with self.assertRaisesRegex(RuntimeError, "status 403"): ensure_collection( "guardian_profiles", {"type": "base", "fields": []}, "token", create_if_missing=True, request_client=fake_client, ) self.assertEqual(calls, ["GET"]) def test_ensure_collection_rejects_500_without_create(self): calls = [] class FakeResponse: def __init__(self, status, payload): self.status = status self._payload = payload def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def read(self): return self._payload.encode("utf-8") def fake_client(request): calls.append(request.get_method()) if request.get_method() == "GET": return FakeResponse(500, "{\"error\":\"boom\"}") if request.get_method() == "POST": raise AssertionError("create must not be attempted on 500") raise AssertionError(f"unexpected method {request.get_method()}") with self.assertRaisesRegex(RuntimeError, "status 500"): ensure_collection( "guardian_profiles", {"type": "base", "fields": []}, "token", create_if_missing=True, request_client=fake_client, ) self.assertEqual(calls, ["GET"]) def test_ensure_collection_rejects_network_failure_without_create(self): calls = [] def fake_client(request): calls.append(request.get_method()) if request.get_method() == "GET": raise urllib.error.URLError("offline") if request.get_method() == "POST": raise AssertionError("create must not be attempted on network failure") raise AssertionError(f"unexpected method {request.get_method()}") with self.assertRaisesRegex(RuntimeError, "status 0"): ensure_collection( "guardian_profiles", {"type": "base", "fields": []}, "token", create_if_missing=True, request_client=fake_client, ) self.assertEqual(calls, ["GET"]) def test_collection_output_is_bare_names_only(self): with patch("setup_auth_family.pb_req", return_value=({"id": "new"}, 201)): buffer = io.StringIO() with redirect_stdout(buffer): from setup_auth_family import create_collection create_collection("guardian_profiles", {"type": "base"}, token="token") self.assertEqual(buffer.getvalue().strip(), "guardian_profiles") if __name__ == "__main__": unittest.main()