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.
		
		
		
		
		
			
		
			
				
					
					
						
							74 lines
						
					
					
						
							2.5 KiB
						
					
					
				
			
		
		
	
	
							74 lines
						
					
					
						
							2.5 KiB
						
					
					
				import json
 | 
						|
from Crypto.PublicKey import RSA
 | 
						|
from pathlib import Path
 | 
						|
 | 
						|
from openpilot.common.params import Params
 | 
						|
from openpilot.system.athena.registration import register, UNREGISTERED_DONGLE_ID
 | 
						|
from openpilot.system.athena.tests.helpers import MockResponse
 | 
						|
from openpilot.system.hardware.hw import Paths
 | 
						|
 | 
						|
 | 
						|
class TestRegistration:
 | 
						|
 | 
						|
  def setup_method(self):
 | 
						|
    # clear params and setup key paths
 | 
						|
    self.params = Params()
 | 
						|
    self.params.clear_all()
 | 
						|
 | 
						|
    persist_dir = Path(Paths.persist_root()) / "comma"
 | 
						|
    persist_dir.mkdir(parents=True, exist_ok=True)
 | 
						|
 | 
						|
    self.priv_key = persist_dir / "id_rsa"
 | 
						|
    self.pub_key = persist_dir / "id_rsa.pub"
 | 
						|
 | 
						|
  def _generate_keys(self):
 | 
						|
    self.pub_key.touch()
 | 
						|
    k = RSA.generate(2048)
 | 
						|
    with open(self.priv_key, "wb") as f:
 | 
						|
      f.write(k.export_key())
 | 
						|
    with open(self.pub_key, "wb") as f:
 | 
						|
      f.write(k.publickey().export_key())
 | 
						|
 | 
						|
  def test_valid_cache(self, mocker):
 | 
						|
    # if all params are written, return the cached dongle id
 | 
						|
    self.params.put("IMEI", "imei")
 | 
						|
    self.params.put("HardwareSerial", "serial")
 | 
						|
    self._generate_keys()
 | 
						|
 | 
						|
    m = mocker.patch("openpilot.system.athena.registration.api_get", autospec=True)
 | 
						|
    dongle = "DONGLE_ID_123"
 | 
						|
    self.params.put("DongleId", dongle)
 | 
						|
    assert register() == dongle
 | 
						|
    assert not m.called
 | 
						|
 | 
						|
  def test_no_keys(self, mocker):
 | 
						|
    # missing pubkey
 | 
						|
    m = mocker.patch("openpilot.system.athena.registration.api_get", autospec=True)
 | 
						|
    dongle = register()
 | 
						|
    assert m.call_count == 0
 | 
						|
    assert dongle == UNREGISTERED_DONGLE_ID
 | 
						|
    assert self.params.get("DongleId", encoding='utf-8') == dongle
 | 
						|
 | 
						|
  def test_missing_cache(self, mocker):
 | 
						|
    # keys exist but no dongle id
 | 
						|
    self._generate_keys()
 | 
						|
    m = mocker.patch("openpilot.system.athena.registration.api_get", autospec=True)
 | 
						|
    dongle = "DONGLE_ID_123"
 | 
						|
    m.return_value = MockResponse(json.dumps({'dongle_id': dongle}), 200)
 | 
						|
    assert register() == dongle
 | 
						|
    assert m.call_count == 1
 | 
						|
 | 
						|
    # call again, shouldn't hit the API this time
 | 
						|
    assert register() == dongle
 | 
						|
    assert m.call_count == 1
 | 
						|
    assert self.params.get("DongleId", encoding='utf-8') == dongle
 | 
						|
 | 
						|
  def test_unregistered(self, mocker):
 | 
						|
    # keys exist, but unregistered
 | 
						|
    self._generate_keys()
 | 
						|
    m = mocker.patch("openpilot.system.athena.registration.api_get", autospec=True)
 | 
						|
    m.return_value = MockResponse(None, 402)
 | 
						|
    dongle = register()
 | 
						|
    assert m.call_count == 1
 | 
						|
    assert dongle == UNREGISTERED_DONGLE_ID
 | 
						|
    assert self.params.get("DongleId", encoding='utf-8') == dongle
 | 
						|
 |