diff options
| -rw-r--r-- | pyagl/services/audiomixer.py | 2 | ||||
| -rw-r--r-- | pyagl/services/base.py | 22 | ||||
| -rw-r--r-- | pyagl/tests/test_geoclue.py | 13 | ||||
| -rw-r--r-- | pyagl/tests/test_network.py | 149 | ||||
| -rw-r--r-- | pyagl/tests/test_weather.py | 6 |
5 files changed, 104 insertions, 88 deletions
diff --git a/pyagl/services/audiomixer.py b/pyagl/services/audiomixer.py index 6e447b7..7d14cc8 100644 --- a/pyagl/services/audiomixer.py +++ b/pyagl/services/audiomixer.py @@ -31,7 +31,7 @@ class AudioMixerService(AGLBaseService): parser.add_argument('--getvolume', help='Get volume level', action='store_true') def __init__(self, ip, port=None, service='agl-service-audiomixer'): - super().__init__(api='audiomixer', ip=ip, port=port, service=service) + super().__init__(api='audiomixer', ip=ip, port=port, service=service, runservice=True) async def subscribe(self, event='volume_changed'): # audio mixer uses 'event' instead 'value', return await self.request('subscribe', {'event': event}) diff --git a/pyagl/services/base.py b/pyagl/services/base.py index 40f7640..320abde 100644 --- a/pyagl/services/base.py +++ b/pyagl/services/base.py @@ -185,12 +185,14 @@ class AGLBaseService: # TODO:handle ssh timeouts, asyncssh does not support it apparently, and connect returns context_manager which # cannot be used with asyncio.wait_for if self.ip == 'localhost' or self.ip == '127.0.0.1': - servicename = check_output(f"systemctl --all | grep {self.service}-- | awk '{{print $1}}'".encode(), shell=True) - servicename = servicename.decode().strip() + servicename = check_output(f"systemctl --all --no-legend | grep {self.service}-- | awk '{{print $1}}'", shell=True) + servicename = servicename.decode().strip().splitlines() + servicename = servicename[-1] if len(servicename) else '' # it is possible that two matching services + # are returned - one for uid 0 as root and 1001 as agl-driver, we want the latter if self.service not in servicename: if runservice: await self.runafmservice() - servicename = await check_output(f"systemctl --all | grep {self.service}-- | awk '{{print $1}}'", + servicename = check_output(f"systemctl --all --no-legend | grep {self.service}-- | awk '{{print $1}}'", shell=True) # retry getting service name after starting servicename = servicename.decode().strip() else: @@ -203,8 +205,8 @@ class AGLBaseService: if runservice: self.logger.warning(f' Trying to start service {servicename}') await self.startsystemdservice(servicename=servicename) - pid = check_output(f'systemctl show --property MainPID --value {servicename}', shell=True) - pid = int(pid.stdout.strip(), 10) + pid = check_output(f'systemctl show --property MainPID --value {servicename}'.encode(), shell=True) + pid = int(pid.decode().strip(), 10) else: return None @@ -221,12 +223,16 @@ class AGLBaseService: else: async with asyncssh.connect(self.ip, username='root') as c: - servicename = await c.run(f"systemctl --all | grep {self.service}-- | awk '{{print $1}}'", check=False) - servicename = servicename.stdout.strip() + servicename = await c.run(f"systemctl --all --no-legend | grep {self.service}-- | awk '{{print $1}}'", check=False) + servicename = servicename.stdout.strip().splitlines() + print(f"matching services: {servicename}") + servicename = servicename[-1] if len(servicename) else '' + # it is possible that two matching services + # are returned - one for uid 0 as root and 1001 as agl-driver, we want the latter if self.service not in servicename: if runservice: await self.runafmservice(connection=c) - servicename = await c.run(f"systemctl --all | grep {self.service}-- | awk '{{print $1}}'", + servicename = await c.run(f"systemctl --all --no-legend | grep {self.service}-- | awk '{{print $1}}'", check=False) # retry getting service name after starting servicename = servicename.stdout.strip() else: diff --git a/pyagl/tests/test_geoclue.py b/pyagl/tests/test_geoclue.py index 6bc6cad..7628042 100644 --- a/pyagl/tests/test_geoclue.py +++ b/pyagl/tests/test_geoclue.py @@ -37,14 +37,17 @@ async def service(): yield svc await svc.websocket.close() - +@pytest.mark.xfail(reason='expected to fail if no internet connection is available or geoclue has no data') @pytest.mark.regular async def test_location(event_loop, service: gcs): msgid = await service.location() resp = await service.afbresponse() + if resp.status == 'failed': + if resp.info == 'No GeoClue instance available': + pytest.xfail(resp.info) assert resp.msgid == msgid assert resp.type == AFBT.RESPONSE - assert resp.status == 'success' + assert resp.status == 'success', resp.info @pytest.mark.regular @@ -52,9 +55,9 @@ async def test_subscribe(event_loop, service: gcs): msgid = await service.subscribe() resp = await service.afbresponse() assert resp.status == 'success' - event = await service.afbresponse() - assert event.type == AFBT.EVENT # subscription immediately emits geoclue event - assert event.api == f'{service.api}/location' + # event = await service.afbresponse() + # assert event.type == AFBT.EVENT # subscription immediately emits geoclue event + # assert event.api == f'{service.api}/location' @pytest.mark.regular diff --git a/pyagl/tests/test_network.py b/pyagl/tests/test_network.py index 2eb4ebc..c0f24fc 100644 --- a/pyagl/tests/test_network.py +++ b/pyagl/tests/test_network.py @@ -39,17 +39,21 @@ async def service(): @pytest.fixture(scope='module') -def expected_available_techs(): +def expected_available_interfaces(): techs = os.environ.get('AGL_AVAILABLE_INTERFACES', 'wifi,ethernet,bluetooth').split(',') return techs +@pytest.fixture(scope='module') +def scannable_interfaces(): + scannable = os.environ.get('AGL_SCANNABLE_INTERFACES', 'wifi,bluetooth').split(',') + @pytest.mark.regular async def test_state(event_loop, service: NS): msgid = await service.state() resp = await service.afbresponse() assert resp.status == 'success', resp.info - assert resp.data == 'online' +# assert resp.data == 'online' @pytest.mark.regular @@ -62,13 +66,71 @@ async def test_global_offline(event_loop, service: NS): resp = await service.afbresponse() assert resp.status == 'success', resp.info +@pytest.mark.regular +async def test_disable_wifi(event_loop, service: NS, expected_available_interfaces): + if 'wifi' not in expected_available_interfaces: + pytest.skip('Skipping disable_technology for "wifi" because it is not expected to be available') + msgid = await service.disable_technology('wifi') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +@pytest.mark.regular +async def test_enable_wifi(event_loop, service: NS, expected_available_interfaces): + if 'wifi' not in expected_available_interfaces: + pytest.skip('Skipping enable_technology for "wifi" because it is not expected to be available') + msgid = await service.enable_technology('wifi') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +@pytest.mark.regular +async def test_enable_bluetooth(event_loop, service: NS, expected_available_interfaces): + if 'bluetooth' not in expected_available_interfaces: + pytest.skip('Skipping enable_technology for "bluetooth" because it is not expected to be available') + msgid = await service.enable_technology('bluetooth') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + @pytest.mark.regular -async def test_scan_services(event_loop, service: NS, expected_available_techs): - scannable_techs = ['wifi', 'bluetooth'] - for t in scannable_techs: - if t in expected_available_techs: - msgid = service.scan_services(technology=t) +async def test_disable_bluetooth(event_loop, service: NS, expected_available_interfaces): + if 'bluetooth' not in expected_available_interfaces: + pytest.skip('Skipping disable_technology for "bluetooth" because it is not expected to be available') + msgid = await service.disable_technology('bluetooth') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +@pytest.mark.regular +async def test_enable_ethernet(event_loop, service: NS, expected_available_interfaces): + if 'ethernet' not in expected_available_interfaces: + pytest.skip('Skipping enable_technology for "ethernet" because it is not expected to be available') + msgid = await service.enable_technology('ethernet') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +@pytest.mark.regular +async def test_disable_ethernet(event_loop, service: NS, expected_available_interfaces): + addr, _ = service.websocket.remote_address + if addr != 'localhost' or addr != '127.0.0.1': + pytest.skip('Skipping this test until it is ran locally on the board, ' + 'presuming it is the only available interface to connect to remotely for testing') + if 'ethernet' not in expected_available_interfaces: + pytest.skip('Skipping disable_technology for "ethernet" because it is not expected to be available') + + msgid = await service.disable_technology('ethernet') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +@pytest.mark.regular +@pytest.mark.dependency(depends=['test_enable_wifi']) +async def test_scan_services(event_loop, service: NS, expected_available_interfaces, scannable_interfaces): + for t in scannable_interfaces: + if t in expected_available_interfaces: + msgid = await service.scan_services(technology=t) resp = await service.afbresponse() assert resp.status == 'success', f'scan_services failed for technology {t} - {resp.info}' @@ -91,18 +153,18 @@ async def test_technologies_verb(event_loop, service: NS): @pytest.mark.regular @pytest.mark.dependency(depends=['test_technologies_verb']) -async def test_expected_existing_technologies(event_loop, service: NS, expected_available_techs): +async def test_expected_existing_technologies(event_loop, service: NS, expected_available_interfaces): msgid = await service.technologies() resp = await service.afbresponse() techs = [t['technology'] for t in resp.data['values']] - for t in expected_available_techs: + for t in expected_available_interfaces: assert t in techs, f'"{t}" technology is expected to be available, but it is not' @pytest.mark.regular @pytest.mark.dependency(depends=['test_expected_existing_technologies']) -async def test_get_property(event_loop, service: NS, expected_available_techs): - for t in expected_available_techs: +async def test_get_property(event_loop, service: NS, expected_available_interfaces): + for t in expected_available_interfaces: msgid = await service.get_property(t) resp = await service.afbresponse() assert resp.status == 'success', resp.info @@ -114,9 +176,10 @@ async def test_get_property(event_loop, service: NS, expected_available_techs): @pytest.mark.regular +@pytest.mark.dependency @pytest.mark.xfail(reason='Expecting this to throw "permission denied" via the API, tethering from connmanctl succeeds') -async def test_enable_wifi_tethering(event_loop, service: NS, expected_available_techs): - if 'wifi' not in expected_available_techs: +async def test_enable_wifi_tethering(event_loop, service: NS, expected_available_interfaces): + if 'wifi' not in expected_available_interfaces: pytest.skip('Skipping wifi tethering test because its not marked as available technology') msgid = await service.set_property('wifi', {'tethering': True, 'TetheringIdentifier': 'AGL', @@ -127,8 +190,8 @@ async def test_enable_wifi_tethering(event_loop, service: NS, expected_available @pytest.mark.regular @pytest.mark.dependency(depends='test_enable_wifi_tethering') -async def test_disable_wifi_tethering(event_loop, service: NS, expected_available_techs): - if 'wifi' not in expected_available_techs: +async def test_disable_wifi_tethering(event_loop, service: NS, expected_available_interfaces): + if 'wifi' not in expected_available_interfaces: pytest.skip('Skipping wifi tethering test because its not marked as available technology') msgid = await service.set_property('wifi', {'tethering': False}) resp = await service.afbresponse() @@ -237,60 +300,4 @@ async def test_unsubscribe_agent(event_loop, service: NS): assert resp.status == 'success', resp.info -@pytest.mark.regular -async def test_enable_wifi(event_loop, service: NS, expected_available_techs): - if 'wifi' not in expected_available_techs: - pytest.skip('Skipping enable_technology for "wifi" because it is not expected to be available') - msgid = await service.enable_technology('wifi') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - - -@pytest.mark.regular -async def test_disable_wifi(event_loop, service: NS, expected_available_techs): - if 'wifi' not in expected_available_techs: - pytest.skip('Skipping disable_technology for "wifi" because it is not expected to be available') - msgid = await service.disable_technology('wifi') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - - -@pytest.mark.regular -async def test_enable_bluetooth(event_loop, service: NS, expected_available_techs): - if 'bluetooth' not in expected_available_techs: - pytest.skip('Skipping enable_technology for "bluetooth" because it is not expected to be available') - msgid = await service.enable_technology('bluetooth') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - - -@pytest.mark.regular -async def test_disable_bluetooth(event_loop, service: NS, expected_available_techs): - if 'bluetooth' not in expected_available_techs: - pytest.skip('Skipping disable_technology for "bluetooth" because it is not expected to be available') - msgid = await service.disable_technology('bluetooth') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - - -@pytest.mark.regular -async def test_enable_ethernet(event_loop, service: NS, expected_available_techs): - if 'ethernet' not in expected_available_techs: - pytest.skip('Skipping enable_technology for "ethernet" because it is not expected to be available') - msgid = await service.enable_technology('ethernet') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - -@pytest.mark.regular -async def test_disable_ethernet(event_loop, service: NS, expected_available_techs): - addr, _ = service.websocket.remote_address - if addr != 'localhost' or addr != '127.0.0.1': - pytest.skip('Skipping this test until it is ran locally on the board, ' - 'presuming it is the only available interface to connect to remotely for testing') - if 'ethernet' not in expected_available_techs: - pytest.skip('Skipping disable_technology for "ethernet" because it is not expected to be available') - - msgid = await service.disable_technology('ethernet') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info diff --git a/pyagl/tests/test_weather.py b/pyagl/tests/test_weather.py index 2c68152..cbf4b64 100644 --- a/pyagl/tests/test_weather.py +++ b/pyagl/tests/test_weather.py @@ -44,7 +44,7 @@ async def test_apikey(event_loop, service: ws): assert resp.msgid == msgid assert resp.data['api_key'] == 'a860fa437924aec3d0360cc749e25f0e' - +@pytest.mark.xfail(reason='expecting this to fail without internet connection') async def test_current_weather(event_loop, service: ws): msgid = await service.current_weather() resp = await service.afbresponse() @@ -70,8 +70,8 @@ async def test_subscribe_weather(event_loop, service: ws): msgid = await service.subscribe(event) resp = await service.afbresponse() assert resp.status == 'success' - eventresp = await service.afbresponse() # immediately emits event after subscription - assert eventresp.api == f'{service.api}/{event}' + # eventresp = await service.afbresponse() # immediately emits event after subscription + # assert eventresp.api == f'{service.api}/{event}' @pytest.mark.dependency(depends=['test_subscribe_weather']) |