ucvl-home-vision/test_journal.py

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'],'第一次修改')