import hashlib import http.client import io import json import secrets import time import unittest from unittest.mock import patch from test_app import app import test_households as fixture from baidu import Baidu, download_url from journal import Journal, CHUNK class JournalTests(unittest.TestCase): stop_server = fixture.HouseholdTests.stop_server request = fixture.HouseholdTests.request def setUp(self): fixture.HouseholdTests.setUp(self) for name, value in [('BAIDU', Baidu(app)), ('JOURNAL', Journal(app))]: p = patch.object(app, name, value); p.start(); self.addCleanup(p.stop) self.actor = self.members[self.aid] def note(self, **changes): return dict(text='今日记事\n第二行', occurredOn='2026-10-04', category='日常', clientId=secrets.token_hex(16), visibility='private', revision=0, mediaIds=[], **changes) def saved(self, user=None, visibility='private'): body = self.note(); body['visibility'] = visibility return app.JOURNAL.save(user or self.actor, body)['id'] def test_private_notes_are_not_readable_by_family_manager_or_superadmin(self): key = self.saved() self.assertEqual(len(app.JOURNAL.snapshot(self.actor, {'mode':'mine'})['items']), 1) for user in (self.owner, self.root, self.other, self.members[self.bid]): self.assertEqual(app.JOURNAL.snapshot(user, {'mode':'person','person':self.actor['id']})['items'], []) self.assertEqual(self.request('/api/journal/action', user, dict(id=key,action='like',liked=True))[0], 404) def test_shared_feed_history_search_and_date_filters_reference_the_same_record(self): key = self.saved(visibility='household') for mode in ('feed','person'): result = app.JOURNAL.snapshot(self.owner, dict(mode=mode,person=self.actor['id'],search='第二行',category='日常',**{'from':'2026-10-01','to':'2026-10-31'})) self.assertEqual(result['items'][0]['id'], key) self.assertEqual(app.JOURNAL.snapshot(self.other, {'mode':'feed'})['total'], 0) self.assertEqual(app.JOURNAL.snapshot(self.owner, {'from':'2026-10-05'})['total'], 0) def test_account_cannot_forge_other_household_or_edit_anothers_note(self): key = self.saved(visibility='household') body = dict(app.get_object('journal_entries', key), text='stolen') self.assertEqual(self.request('/api/journal', self.owner, body)[0], 403) self.assertEqual(self.request('/api/journal', self.actor, self.note(), home=self.bid)[0], 403) def test_creation_retry_and_edit_conflict(self): body = self.note(); first = app.JOURNAL.save(self.actor, body); second = app.JOURNAL.save(self.actor, body) self.assertEqual(first['id'], second['id']); self.assertTrue(second['duplicate']) old = app.get_object('journal_entries', first['id']) self.assertEqual(self.request('/api/journal', self.actor, dict(old,text='修改'))[0], 200) self.assertEqual(self.request('/api/journal', self.actor, old)[0], 409) def test_like_set_is_idempotent_and_privacy_change_removes_all_interactions(self): key = self.saved(visibility='household'); data = dict(id=key,action='like',liked=True) for _ in range(2): self.assertEqual(self.request('/api/journal/action', self.owner, data)[0], 200) row = app.JOURNAL.snapshot(self.actor, {'mode':'mine'})['items'][0]; self.assertEqual(row['likes'],1) app.JOURNAL.save(self.actor, dict(app.get_object('journal_entries', key),visibility='private')) self.assertEqual(self.request('/api/journal/action', self.owner, dict(data,liked=False))[0],404) def test_comments_allow_author_or_entry_owner_to_remove_but_no_other_member(self): key = self.saved(visibility='household'); cid=secrets.token_hex(16) data=dict(id=key,action='comment',clientId=cid,text='你好') for _ in range(2): self.assertEqual(self.request('/api/journal/action', self.owner, data)[0],200) self.assertEqual(len(app.objects('journal_comments')),1) delete=dict(id=key,action='delete-comment',commentId=cid) self.assertEqual(self.request('/api/journal/action', self.backup, delete)[0],403) self.assertEqual(self.request('/api/journal/action', self.actor, delete)[0],200) def test_unready_or_foreign_media_cannot_be_attached_and_unlinked_media_is_private(self): media=dict(id=secrets.token_hex(8),familyId=self.aid,ownerId=self.actor['id'],status='ready',size=8,name='x.png',mime='image/png',at='now') app.save_object('journal_media', media) body=self.note();body['mediaIds']=[media['id']] self.assertEqual(self.request('/api/journal', self.owner, body)[0],403) with self.assertRaises(app.Problem): app.JOURNAL.media(self.owner, media['id']) body['visibility']='household';key=app.JOURNAL.save(self.actor,body)['id'] self.assertEqual(app.JOURNAL.media(self.owner,media['id'])['id'],media['id']) app.JOURNAL.action(self.actor,dict(id=key,action='delete',revision=1)) with self.assertRaises(app.Problem): app.JOURNAL.media(self.owner,media['id']) def test_upload_sends_bounded_blocks_without_writing_any_media_file(self): raw=b'\x89PNG\r\n\x1a\n'+b'a'*(CHUNK+20) before=set(app.DATA.rglob('*')) with patch.object(app.BAIDU,'token',return_value='fake'), patch.object(app.BAIDU,'prepare',return_value={'uploadid':'u','block_list':[0]}), patch.object(app.BAIDU,'send_block') as send: task=app.JOURNAL.begin_upload(self.actor,dict(name='x.png',size=len(raw),hashes=[hashlib.md5(raw).hexdigest()])) first=app.JOURNAL.chunk(self.actor,task['id'],0,raw[:CHUNK]);self.assertEqual(first['received'],CHUNK) app.JOURNAL.chunk(self.actor,task['id'],0,raw[:CHUNK]) # idempotent partial retry app.JOURNAL.chunk(self.actor,task['id'],CHUNK,raw[CHUNK:]);send.assert_called_once() self.assertEqual(send.call_args.args[3],raw) self.assertEqual(set(app.DATA.rglob('*')), before) self.assertFalse(app.JOURNAL.buffers) def test_resume_after_restart_uses_cloud_missing_blocks(self): data=dict(name='v.mp4',size=4194305,hashes=['a'*32,'b'*32]) with patch.object(app.BAIDU,'token',return_value='fake'), patch.object(app.BAIDU,'prepare',return_value={'uploadid':'u','block_list':[1]}): task=app.JOURNAL.begin_upload(self.actor,data) self.assertEqual(task['received'],4194304) again=app.JOURNAL.begin_upload(self.actor,dict(data,id=task['id'])) self.assertEqual(again['id'],task['id']) def test_raw_upload_api_auth_and_foreign_user(self): raw=b'\x89PNG\r\n\x1a\nabc' with patch.object(app.BAIDU,'token',return_value='fake'), patch.object(app.BAIDU,'prepare',return_value={'uploadid':'u','block_list':[0]}): task=app.JOURNAL.begin_upload(self.actor,dict(name='x.png',size=len(raw),hashes=[hashlib.md5(raw).hexdigest()])) c=http.client.HTTPConnection(*self.server.server_address) c.request('POST','/api/journal/chunk?id='+task['id']+'&offset=0',raw,{'Cookie':'vision='+self.tokens[self.owner['id']]}) r=c.getresponse();r.read();c.close();self.assertEqual(r.status,403) def test_media_and_oauth_metadata_do_not_reveal_tokens_or_remote_paths(self): self.assertEqual(self.request('/api/baidu/config',self.owner)[0],403) self.assertEqual(self.request('/api/baidu/connect',self.owner,{})[0],403) self.assertEqual(self.request('/api/baidu/disconnect',self.owner,{})[0],403) body=app.JOURNAL.snapshot(self.actor,{'mode':'mine'}) self.assertNotIn('secretKey',json.dumps(body));self.assertNotIn('access_token',json.dumps(body)) def test_media_http_requires_visibility_and_preserves_range_without_cloud_redirect(self): media=dict(id=secrets.token_hex(8),familyId=self.aid,ownerId=self.actor['id'],status='ready', size=10,name='clip.mp4',mime='video/mp4',at='now',fsId='123',remotePath='/private/clip.mp4') app.save_object('journal_media',media) key=app.JOURNAL.save(self.actor,dict(self.note(),mediaIds=[media['id']]))['id'] path='/media/journal?id='+media['id']+'&family='+self.aid def fetch(user): c=http.client.HTTPConnection(*self.server.server_address) headers={'Range':'bytes=2-4'} if user: headers['Cookie']='vision='+self.tokens[user['id']] c.request('GET',path,headers=headers);r=c.getresponse();result=(r.status,dict(r.getheaders()),r.read());c.close();return result with patch.object(app.BAIDU,'open_file') as opened: self.assertEqual(fetch(None)[0],401) self.assertEqual(fetch(self.owner)[0],404) opened.assert_not_called() app.JOURNAL.save(self.actor,dict(app.get_object('journal_entries',key),visibility='household')) response=io.BytesIO(b'234');response.status=206 response.headers={'Content-Length':'3','Content-Range':'bytes 2-4/10'};opened.return_value=response status,headers,body=fetch(self.owner) self.assertEqual((status,body),(206,b'234'));self.assertEqual(headers['Cache-Control'],'no-store') self.assertEqual(headers['Content-Range'],'bytes 2-4/10');self.assertNotIn('Location',headers) self.assertEqual(opened.call_args.args[2],'bytes=2-4') def config(self): return dict(appKey='fixtureKey',secretKey='fixtureSecret',appFolder='测试',redirectUri='https://example.org/api/baidu/callback',revision=0) def test_config_revisions_callback_validation_and_secret_not_returned(self): body=app.BAIDU.configure(self.config(),self.root) self.assertTrue(body['hasSecret']);self.assertNotIn('secretKey',body) with self.assertRaises(app.Problem): app.BAIDU.configure(self.config(),self.root) for address in ('http://example.org/api/baidu/callback','https://example.org/other','https://a:b@example.org/api/baidu/callback'): with self.assertRaises(app.Problem): app.BAIDU.configure(dict(self.config(),revision=1,redirectUri=address),self.root) def test_oauth_state_cookie_session_and_replay_protection(self): app.BAIDU.configure(self.config(),self.root) result,cookie=app.BAIDU.begin(self.root,'media',self.tokens[self.root['id']],'example.org') state=cookie.split(';')[0].split('=')[1] with self.assertRaises(app.Problem): app.BAIDU.callback(dict(state=state,code='x'),'') token=dict(access_token='test-token',refresh_token='test-refresh',expires=time.time()+1000) with patch.object(app.BAIDU,'token_response',return_value=token),patch.object(app.BAIDU,'json_request',return_value={'uk':123,'baidu_name':'测试网盘'}): app.BAIDU.callback(dict(state=state,code='x'),cookie) with self.assertRaises(app.Problem): app.BAIDU.callback(dict(state=state,code='x'),cookie) self.assertTrue(app.BAIDU.status(self.actor)['connected']) self.assertNotIn('test-token',json.dumps(app.BAIDU.status(self.actor))) def test_logged_out_or_demoted_oauth_cannot_bind_account(self): app.BAIDU.configure(self.config(),self.root) _,cookie=app.BAIDU.begin(self.root,'media',self.tokens[self.root['id']],'example.org') state=cookie.split(';')[0].split('=')[1];app.DB.execute('DELETE FROM sessions');app.DB.commit() with self.assertRaises(app.Problem): app.BAIDU.callback(dict(state=state,code='x'),cookie) def test_download_redirect_targets_are_constrained(self): for address in ('https://d.pcs.baidu.com/file/a','https://x.baidupcs.com/a'): self.assertEqual(download_url(address),address) for address in ('http://d.pcs.baidu.com/a','https://evil.test/a','https://baidu.com.evil.test/a','https://user@baidu.com/a','https://127.0.0.1/a'): with self.assertRaises(ValueError): download_url(address) def test_removed_attachment_and_invalid_text_never_publish(self): bad=self.note();bad['text']=[] self.assertEqual(self.request('/api/journal',self.actor,bad)[0],400) bad=self.note();bad['occurredOn']='2026-99-99' self.assertEqual(self.request('/api/journal',self.actor,bad)[0],400) self.assertEqual(self.request('/api/journal',self.actor,dict(self.note(),visibility='public'))[0],400) def test_household_profile_revision_preserves_independent_leadership(self): home=app.get_object('households',self.aid) data=dict(home,name='第一次修改') self.assertEqual(self.request('/api/households',self.root,data)[0],200) self.assertEqual(self.request('/api/households',self.root,data)[0],409) self.assertEqual(app.get_object('households',self.aid)['name'],'第一次修改') def test_attachment_trash_restore_and_existing_reference_remain_visible(self): m=dict(id=secrets.token_hex(8),familyId=self.aid,ownerId=self.actor['id'],status='ready',size=8,name='x.png',mime='image/png',at='now') app.save_object('journal_media',m) key=app.JOURNAL.save(self.actor,dict(self.note(),visibility='household',mediaIds=[m['id']]))['id'] self.assertEqual(self.request('/api/journal/remove-upload',self.actor,{'id':m['id']})[0],200) snap=app.JOURNAL.snapshot(self.actor,{'mode':'mine'});self.assertEqual(snap['uploads'],[]);self.assertEqual(snap['trash'][0]['id'],m['id']) self.assertEqual(app.JOURNAL.media(self.owner,m['id'])['id'],m['id']) self.assertEqual(self.request('/api/journal',self.actor,dict(self.note(),mediaIds=[m['id']]))[0],409) self.assertEqual(self.request('/api/journal',self.actor,app.get_object('journal_entries',key))[0],200) for user in (self.owner,self.other): self.assertIn(self.request('/api/journal/restore-upload',user,{'id':m['id']})[0],(403,404)) self.assertEqual(app.JOURNAL.snapshot(user,{})['trash'],[]) self.assertEqual(self.request('/api/journal/restore-upload',self.actor,{'id':m['id']})[0],200) self.assertEqual(app.JOURNAL.snapshot(self.actor,{})['trash'],[]) def test_deleted_partial_upload_rejects_chunk_and_finish_and_can_restore(self): m=dict(id=secrets.token_hex(8),familyId=self.aid,ownerId=self.actor['id'],status='receiving',size=8,name='x.png',mime='image/png',at='now',received=0) app.save_object('journal_media',m);app.JOURNAL.remove_upload(self.actor,m['id']) for action in (lambda:app.JOURNAL.chunk(self.actor,m['id'],0,b'123'),lambda:app.JOURNAL.finish_upload(self.actor,m['id'])): with self.assertRaises(app.Problem):action() app.JOURNAL.restore_upload(self.actor,m['id']);self.assertFalse(app.get_object('journal_media',m['id'])['trashedAt']) def test_backup_readback_hash_check_and_constrained_cloud_delete(self): data=b'fixture sqlite';stream=io.BytesIO(data) with patch.object(app.BAIDU,'open_file',return_value=stream): app.BAIDU.verify_backup('/test','1',len(data),hashlib.sha256(data).hexdigest()) with patch.object(app.BAIDU,'open_file',return_value=io.BytesIO(b'wrong')): with self.assertRaises(app.Problem):app.BAIDU.verify_backup('/test','1',len(data),'wrong') root='/apps/测试/数据库备份/'+'a'*16;name='vision-'+'a'*16+'-20260801T120000000000Z.sqlite';valid=root+'/'+name with patch.object(app.BAIDU,'public_config',return_value={'appFolder':'测试'}),patch.object(app.BAIDU,'request',return_value={'info':[{'path':valid,'errno':0}]}) as request: app.BAIDU.delete_backup_file(valid) for path in (root,root+'/../'+name,root+'/other.png',root+'/'+name+'/private.png','/apps/测试/资料/'+name): with self.assertRaises(app.Problem):app.BAIDU.delete_backup_file(path) self.assertEqual(request.call_count,1) def test_person_structured_places_preserve_legacy_and_survive_old_clients(self): data=dict(name='测试人物',birthplace='旧称原记录',nativeProvince='山东省',nativeCity='烟台市',nativeDistrict='莱阳市',nativeTown='测试镇',nativeCommunity='测试村',birthCountry='中国',birthCity='济南市') person=app.FAMILY.person(data,None) self.assertEqual(person['nativeCity'],'烟台市');self.assertEqual(person['birthplace'],'旧称原记录') saved=app.FAMILY.person({'name':'新姓名','birthplace':'旧称原记录'},person) self.assertEqual(saved['nativeCommunity'],'测试村');self.assertEqual(saved['birthCity'],'济南市') with self.assertRaises(app.Problem):app.FAMILY.person(dict(data,nativeCity=[]),person)