Add isolated Athena Deck prototype with demo control and live hardware telemetry
This commit is contained in:
@@ -0,0 +1,45 @@
|
||||
import json
|
||||
import threading
|
||||
import unittest
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from unittest.mock import patch
|
||||
from server import Server, HardwareProvider
|
||||
|
||||
class Tests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.server=Server(0)
|
||||
self.thread=threading.Thread(target=self.server.serve_forever)
|
||||
self.thread.start()
|
||||
self.url=f'http://127.0.0.1:{self.server.server_port}'
|
||||
def tearDown(self):
|
||||
self.server.shutdown()
|
||||
self.server.demo.stop()
|
||||
self.server.server_close()
|
||||
self.thread.join()
|
||||
def request(self,path,method='GET',headers=None):
|
||||
req=urllib.request.Request(self.url+'/api/v1/'+path,method=method,headers=headers or {})
|
||||
with urllib.request.urlopen(req) as r:return json.load(r)
|
||||
def test_lifecycle(self):
|
||||
self.assertEqual(self.request('status')['demo']['state'],'stopped')
|
||||
a=self.request('demo/start','POST',{'X-Athena-Deck':'1'})
|
||||
self.assertTrue(a['reachable'])
|
||||
self.assertEqual(a['pid'],self.request('demo/start','POST',{'X-Athena-Deck':'1'})['pid'])
|
||||
child=self.server.demo.process
|
||||
self.assertEqual(self.request('demo/stop','POST',{'X-Athena-Deck':'1'})['state'],'stopped')
|
||||
self.assertIsNotNone(child.poll())
|
||||
self.assertFalse(self.request('demo/stop','POST',{'X-Athena-Deck':'1'})['reachable'])
|
||||
def test_control_guard(self):
|
||||
for headers in ({},{'X-Athena-Deck':'1','Origin':'http://evil.invalid'}):
|
||||
with self.assertRaises(urllib.error.HTTPError) as e:self.request('demo/start','POST',headers)
|
||||
self.assertEqual(e.exception.code,403)
|
||||
with self.assertRaises(urllib.error.HTTPError) as e:self.request('models/start','POST',{'X-Athena-Deck':'1'})
|
||||
self.assertEqual(e.exception.code,404)
|
||||
def test_unavailable_hardware(self):
|
||||
with patch('server.subprocess.run',side_effect=OSError()):
|
||||
result=HardwareProvider().snapshot()
|
||||
self.assertFalse(result['available'])
|
||||
self.assertEqual(result['gpus'],[])
|
||||
self.assertIsNone(result['sampled_at'])
|
||||
|
||||
if __name__=='__main__':unittest.main()
|
||||
Reference in New Issue
Block a user