You can not select more than 25 topics
			Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
		
		
		
		
		
			
		
			
				
					
					
						
							82 lines
						
					
					
						
							2.5 KiB
						
					
					
				
			
		
		
	
	
							82 lines
						
					
					
						
							2.5 KiB
						
					
					
				#!/usr/bin/env python3
 | 
						|
import os
 | 
						|
import shutil
 | 
						|
import subprocess
 | 
						|
import tempfile
 | 
						|
import time
 | 
						|
import unittest
 | 
						|
 | 
						|
from common.basedir import BASEDIR
 | 
						|
 | 
						|
UPDATER_PATH = os.path.join(BASEDIR, "installer/updater")
 | 
						|
UPDATER = os.path.join(UPDATER_PATH, "updater")
 | 
						|
UPDATE_MANIFEST = os.path.join(UPDATER_PATH, "update.json")
 | 
						|
 | 
						|
 | 
						|
class TestUpdater(unittest.TestCase):
 | 
						|
 | 
						|
  @classmethod
 | 
						|
  def setUpClass(cls):
 | 
						|
    # test that the updater builds
 | 
						|
    cls.assertTrue(f"cd {UPDATER_PATH} && make clean && make", "updater failed to build")
 | 
						|
 | 
						|
    # restore the checked-in version, since that's what actually runs on devices
 | 
						|
    os.system(f"git reset --hard {UPDATER_PATH}")
 | 
						|
 | 
						|
  def setUp(self):
 | 
						|
    self._clear_dir()
 | 
						|
 | 
						|
  def tearDown(self):
 | 
						|
    self._clear_dir()
 | 
						|
 | 
						|
  def _clear_dir(self):
 | 
						|
    if os.path.isdir("/data/neoupdate"):
 | 
						|
      shutil.rmtree("/data/neoupdate")
 | 
						|
 | 
						|
  def _assert_ok(self, cmd, msg=None):
 | 
						|
    self.assertTrue(os.system(cmd) == 0, msg)
 | 
						|
 | 
						|
  def _assert_fails(self, cmd):
 | 
						|
    self.assertFalse(os.system(cmd) == 0)
 | 
						|
 | 
						|
  def test_background_download(self):
 | 
						|
    self._assert_ok(f"{UPDATER} bgcache 'file://{UPDATE_MANIFEST}'")
 | 
						|
 | 
						|
  def test_background_download_bad_manifest(self):
 | 
						|
    # update with bad manifest should fail
 | 
						|
    with tempfile.NamedTemporaryFile(mode="w", suffix=".json") as f:
 | 
						|
      f.write("{}")
 | 
						|
      self._assert_fails(f"{UPDATER} bgcache 'file://{f.name}'")
 | 
						|
 | 
						|
  def test_cache_resume(self):
 | 
						|
    self._assert_ok(f"{UPDATER} bgcache 'file://{UPDATE_MANIFEST}'")
 | 
						|
    # a full download takes >1m, but resuming from fully cached should only be a few seconds
 | 
						|
    start_time = time.monotonic()
 | 
						|
    self._assert_ok(f"{UPDATER} bgcache 'file://{UPDATE_MANIFEST}'")
 | 
						|
    self.assertLess(time.monotonic() - start_time, 10)
 | 
						|
 | 
						|
  # make sure we can recover from corrupt downloads
 | 
						|
  def test_recover_from_corrupt(self):
 | 
						|
    # download the whole update
 | 
						|
    self._assert_ok(f"{UPDATER} bgcache 'file://{UPDATE_MANIFEST}'")
 | 
						|
 | 
						|
    # write some random bytes
 | 
						|
    for f in os.listdir("/data/neoupdate"):
 | 
						|
      with open(os.path.join("/data/neoupdate", f), "ab") as f:
 | 
						|
        f.write(b"\xab"*20)
 | 
						|
 | 
						|
    # this attempt should fail, then it unlinks
 | 
						|
    self._assert_fails(f"{UPDATER} bgcache 'file://{UPDATE_MANIFEST}'")
 | 
						|
 | 
						|
    # now it should pass
 | 
						|
    self._assert_ok(f"{UPDATER} bgcache 'file://{UPDATE_MANIFEST}'")
 | 
						|
 | 
						|
  # simple test that the updater doesn't crash in UI mode
 | 
						|
  def test_ui_init(self):
 | 
						|
    with subprocess.Popen(UPDATER) as proc:
 | 
						|
      time.sleep(5)
 | 
						|
      self.assertTrue(proc.poll() is None)
 | 
						|
      proc.terminate()
 | 
						|
 | 
						|
if __name__ == "__main__":
 | 
						|
  unittest.main()
 | 
						|
 |