"""Real storage and HTTP regressions for the universal five-generation boundary.""" import hashlib import http.client import json import secrets import sqlite3 import tempfile import threading import time import unittest from pathlib import Path from unittest.mock import patch from test_app import app from family_permissions import generations class FamilyPermissionTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) db = sqlite3.connect(':memory:', check_same_thread=False); db.row_factory = sqlite3.Row self.addCleanup(db.close) db.executescript('CREATE TABLE settings (key TEXT PRIMARY KEY,value TEXT NOT NULL);' 'CREATE TABLE sessions (hash TEXT PRIMARY KEY,expires REAL NOT NULL);' 'CREATE TABLE audit (id INTEGER PRIMARY KEY,at TEXT,action TEXT,name TEXT);') for table in ('sites', 'assets', 'recorders', 'cameras'): db.execute('CREATE TABLE '+table+' (id TEXT PRIMARY KEY,body TEXT NOT NULL)') for key, value in [('DB', db), ('DATA', Path(self.temp.name)), ('write_media_config', lambda: None)]: p = patch.object(app, key, value); p.start(); self.addCleanup(p.stop) app.initialize() self.actor = app.ACCOUNTS.save(dict(username='admin', password='fixture-only', role='admin'), None) self.home = self.actor['familyId'] self.people = {} for generation in range(5, 40): person = dict(app.FAMILY.person(dict(name='第'+str(generation)+'代'), None), familyId=self.home) app.save_object('people', person); self.people[generation] = person for generation in range(5, 39): edge = dict(id=secrets.token_hex(8), fromId=self.people[generation]['id'], toId=self.people[generation+1]['id'], kind='parent', lineage='unspecified', active=True, note='', familyId=self.home) app.save_object('family_links', edge) self.actor = app.ACCOUNTS.save(dict(self.actor, personId=self.people[32]['id']), self.actor) def save(self, kind, item, **extra): return app.FAMILY.save(kind, dict(item, revision=app.setting('familyRevision', 0), **extra), self.actor) def test_all_roles_can_edit_fifth_but_not_sixth_or_ancient_generation(self): for role in ('admin', 'family_admin', 'member'): self.actor = dict(self.actor, role=role, familyAccess='edit') for generation in (27, 32, 37): self.save('person', dict(self.people[generation], note=role)) for generation in (5, 26, 38): with self.subTest(role=role, generation=generation), self.assertRaises(app.Problem) as caught: self.save('person', dict(self.people[generation], note='denied')) self.assertEqual(caught.exception.status, 403) self.assertEqual(app.get_object('people', self.people[generation]['id'])['note'], '') def test_view_contains_all_generations_and_reference_cannot_change_scope(self): view = app.FAMILY.snapshot(self.actor) self.assertEqual(len(view['people']), 35) self.assertEqual(view['editScope']['levels'][self.people[5]['id']], -27) self.assertNotIn(self.people[5]['id'], view['editScope']['editablePersonIds']) self.assertEqual(len(view['editScope']['editablePersonIds']), 11) with self.assertRaises(app.Problem): app.FAMILY.save('person', dict(self.people[5], note='forged', revision=0, selfId=self.people[5]['id']), dict(self.actor, personId=self.people[5]['id'])) def test_new_relative_is_atomic_and_sixth_leaves_no_orphan(self): before = len(app.objects('people')) for relative, kind in ((27, 'parent'), (37, 'child')): with self.assertRaises(app.Problem) as caught: self.save('person', dict(name='Outside'), relativeId=self.people[relative]['id'], relativeKind=kind) self.assertEqual(caught.exception.status, 403) self.assertEqual(len(app.objects('people')), before) result = self.save('person', dict(name='A sibling'), relativeId=self.people[31]['id'], relativeKind='child') self.assertEqual(app.FAMILY.snapshot(self.actor)['editScope']['levels'][result['item']['id']], 0) self.assertTrue(any(e['toId'] == result['item']['id'] for e in app.objects('family_links'))) def test_relationship_rewrite_and_disconnect_cannot_launder_scope(self): with self.assertRaises(app.Problem): self.save('link', dict(fromId=self.people[32]['id'], toId=self.people[5]['id'], kind='spouse')) edge = next(e for e in app.objects('family_links') if e['fromId'] == self.people[30]['id']) with self.assertRaises(app.Problem): self.save('link', dict(edge, active=False)) self.assertTrue(app.get_object('family_links', edge['id'])['active']) with self.assertRaises(app.Problem): self.save('link', dict(fromId=self.people[28]['id'], toId=self.people[30]['id'], kind='parent')) def test_old_event_cannot_be_retargeted_to_self_to_gain_write_access(self): event = dict(app.FAMILY.event(dict(title='Old ancestor', eventDate='1900-01-01', body='untouched', personIds=[self.people[5]['id']], status='published'), None), familyId=self.home) app.save_object('family_events', event) with self.assertRaises(app.Problem): self.save('event', dict(event, personIds=[self.people[32]['id']], body='denied')) self.assertEqual(app.get_object('family_events', event['id'])['body'], 'untouched') self.assertIn(event['id'], [e['id'] for e in app.FAMILY.snapshot(self.actor)['events']]) def test_history_restore_cannot_change_far_person(self): person = self.people[5] app.DB.execute('INSERT INTO family_changes VALUES (?,?,?,?,?,?,?,?)', (1, 'person', person['id'], None, json.dumps(dict(person, note='old text')), 'fixture', '2026-10-01', None)) app.DB.commit() with self.assertRaises(app.Problem) as caught: app.FAMILY.restore(dict(changeRevision=1, side='after', revision=0), self.actor) self.assertEqual(caught.exception.status, 403) self.assertEqual(app.get_object('people', person['id'])['note'], '') def test_binding_cannot_be_cleared_or_changed_even_by_super_admin(self): for person in ('', self.people[5]['id']): with self.assertRaises(app.Problem) as caught: app.ACCOUNTS.save(dict(self.actor, personId=person), self.actor) self.assertEqual(caught.exception.status, 403) def test_unbound_account_can_create_only_its_own_starting_record(self): account = app.ACCOUNTS.save(dict(username='unbound', password='fixture-only', familyAccess='edit'), self.actor) self.assertFalse(app.FAMILY.snapshot(account)['canEdit']) with self.assertRaises(app.Problem): app.FAMILY.save('person', dict(self.people[32], note='denied', revision=0), account) result = app.FAMILY.save('person', dict(name='New self', asSelf=True, revision=0), account) self.assertEqual(app.get_object('users', account['id'])['personId'], result['item']['id']) with self.assertRaises(app.Problem): app.FAMILY.save('person', dict(name='Another self', asSelf=True, revision=1), account) def test_conflicting_generations_fail_closed(self): edge = dict(id=secrets.token_hex(8), fromId=self.people[27]['id'], toId=self.people[29]['id'], kind='parent', active=True, familyId=self.home) app.save_object('family_links', edge) self.assertTrue(app.FAMILY.snapshot(self.actor)['editScope']['conflict']) self.assertFalse(app.FAMILY.snapshot(self.actor)['canEdit']) with self.assertRaises(app.Problem): self.save('person', dict(self.people[32], note='denied')) def test_http_direct_write_is_denied_for_super_admin(self): token = 'a'*64 app.DB.execute('INSERT INTO sessions VALUES (?,?,?)', (hashlib.sha256(token.encode()).hexdigest(), time.time()+60, self.actor['id'])) app.DB.commit() server = app.ThreadingHTTPServer(('127.0.0.1', 0), app.Handler) thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start() try: c = http.client.HTTPConnection(*server.server_address, timeout=3) c.request('POST', '/api/family/person', json.dumps(dict(self.people[5], note='denied', revision=0)), {'Cookie': 'vision='+token, 'Content-Type': 'application/json'}) response = c.getresponse(); response.read(); c.close() self.assertEqual(response.status, 403) finally: server.shutdown(); server.server_close(); thread.join() def test_offsets_are_generation_difference_not_number_of_clicks(self): spouse = dict(app.FAMILY.person({'name':'spouse'}, None), familyId=self.home) app.save_object('people', spouse) app.save_object('family_links', dict(id=secrets.token_hex(8), fromId=self.people[32]['id'], toId=spouse['id'], kind='spouse', active=True, familyId=self.home)) levels, conflict = generations(app.objects('people'), app.objects('family_links'), self.people[32]['id']) self.assertFalse(conflict); self.assertEqual(levels[spouse['id']], 0) if __name__ == '__main__': unittest.main()