cheng
74 lines
3.6 KiB
Python
74 lines
3.6 KiB
Python
import json
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from unittest import TestCase
|
|
from unittest.mock import patch
|
|
from station import operator_camera_view as preview
|
|
|
|
|
|
class PreviewCacheTest(TestCase):
|
|
def tearDown(self):
|
|
preview.watch(set(),set())
|
|
preview._PREVIEW.clear();preview._META.clear()
|
|
|
|
def test_switch_and_stale_worker_clear_frames(self):
|
|
preview.watch({'left_color'},set())
|
|
preview._PREVIEW['left_color']=(b'123',1,1)
|
|
preview.watch({'right_color'},set())
|
|
self.assertNotIn('left_color',preview.previews())
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
preview._PREVIEW['right_color']=(b'123',1,1)
|
|
payload={'frames':{},'exported_monotonic':time.monotonic()-10,'meta':{},'ages':{}}
|
|
(Path(folder)/'meta.json').write_text(json.dumps(payload))
|
|
with patch.object(preview,'_BRIDGE_DIR',Path(folder)):
|
|
preview._ingest()
|
|
self.assertEqual(preview.previews(),{})
|
|
self.assertIn('未更新',preview.state()['error'])
|
|
|
|
def test_unchanged_shared_frame_is_not_copied_again(self):
|
|
from types import SimpleNamespace
|
|
preview.watch({'left_color'},set())
|
|
header=bytearray(preview._SHM_HEAD.size)
|
|
preview._SHM_HEAD.pack_into(header,0,2,1280,720,1280*720*3)
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
payload={'frames':{'left_color':{'name':'owned_test','store':'left_color','kind':'rgb'}},
|
|
'meta':{},'ages':{'left_color':0},'exported_monotonic':time.monotonic()}
|
|
(Path(folder)/'meta.json').write_text(json.dumps(payload))
|
|
with patch.object(preview,'_BRIDGE_DIR',Path(folder)), \
|
|
patch.object(preview,'_SHM_SEQ',{'owned_test':2}), \
|
|
patch.object(preview,'_attach',return_value=SimpleNamespace(buf=header)), \
|
|
patch.object(preview,'_shm_read',side_effect=AssertionError('unchanged frame copied')):
|
|
preview._ingest()
|
|
|
|
def test_full_hd_is_preserved_and_larger_frames_fit_transport(self):
|
|
import numpy as np
|
|
image=np.zeros((1080,1920,3),np.uint8)
|
|
self.assertEqual(preview.rgb_preview(image,bgr=False).shape,image.shape)
|
|
portrait=preview.rgb_preview(np.zeros((1280,720,3),np.uint8),bgr=False)
|
|
self.assertLessEqual(portrait.shape[0],1080)
|
|
self.assertLessEqual(portrait.nbytes,preview._SHM_MAX)
|
|
|
|
def test_refresh_worker_exit_does_not_kill_preview(self):
|
|
import os,sys,subprocess,threading
|
|
real_popen=subprocess.Popen
|
|
spawning=[]
|
|
def popen(argv,**kwargs):
|
|
spawning.append(threading.current_thread().name)
|
|
command=[sys.executable,'-m','station.child_process',str(os.getpid()),
|
|
sys.executable,'-c','import time;time.sleep(20)']
|
|
proc=real_popen(command,**kwargs)
|
|
time.sleep(.2) # Allow the child to install its parent-death watch.
|
|
return proc
|
|
with patch.object(preview,'stream_targets',return_value=[{'id':'fake','label':'fake'}]), \
|
|
patch.object(preview.subprocess,'Popen',side_effect=popen), \
|
|
patch.object(preview,'_poll_loop',side_effect=lambda:preview._STOP.wait(5)):
|
|
worker=threading.Thread(target=preview.start,name='short-refresh-worker')
|
|
worker.start();worker.join(3)
|
|
try:
|
|
self.assertFalse(worker.is_alive())
|
|
self.assertEqual(spawning,['cam-poll'])
|
|
time.sleep(.3)
|
|
self.assertIsNone(preview._BRIDGE.poll())
|
|
finally:preview.stop()
|