Fix flake8 warnings

agrirouter/utils/type_url.py

@@ -14,32 +14,27 @@
 
 class TypeUrl:
     prefix = "types.agrirouter.com/"
+    commands = (
+        Messages,
+        ListEndpointsResponse,
+        HeaderQueryResponse,
+        MessageQueryResponse,
+        MessageDelete,
+        MessageConfirm,
+        OnboardingResponse,
+        OnboardingRequest,
+        CapabilitySpecification,
+        Subscription,
+        MessageQuery,
+        ListEndpointsQuery,
+    )
 
     @classmethod
     def get_type_url(cls, class_):
-        if class_ == Messages:
-            return cls.prefix + Messages.DESCRIPTOR.full_name
-        elif class_ == ListEndpointsResponse:
-            return cls.prefix + ListEndpointsResponse.DESCRIPTOR.full_name
-        elif class_ == HeaderQueryResponse:
-            return cls.prefix + HeaderQueryResponse.DESCRIPTOR.full_name
-        elif class_ == MessageQueryResponse:
-            return cls.prefix + MessageQueryResponse.DESCRIPTOR.full_name
-        elif class_ == MessageDelete:
-            return cls.prefix + MessageDelete.DESCRIPTOR.full_name
-        elif class_ == MessageConfirm:
-            return cls.prefix + MessageConfirm.DESCRIPTOR.full_name
-        elif class_ == OnboardingResponse:
-            return cls.prefix + OnboardingResponse.DESCRIPTOR.full_name
-        elif class_ == OnboardingRequest:
-            return cls.prefix + OnboardingRequest.DESCRIPTOR.full_name
-        elif class_ == CapabilitySpecification:
-            return cls.prefix + CapabilitySpecification.DESCRIPTOR.full_name
-        elif class_ == Subscription:
-            return cls.prefix + Subscription.DESCRIPTOR.full_name
-        elif class_ == MessageQuery:
-            return cls.prefix + MessageQuery.DESCRIPTOR.full_name
-        elif class_ == ListEndpointsQuery:
-            return cls.prefix + ListEndpointsQuery.DESCRIPTOR.full_name
-        else:
+        return TypeUrl.get_command(class_)
+
+    @classmethod
+    def get_command(cls, class_) -> str:
+        if class_ not in cls.commands:
             raise TypeUrlNotFoundError(f"The {class_} type url not found")
+        return cls.prefix + class_.DESCRIPTOR.full_name

setup.py

@@ -12,27 +12,27 @@
     description="""This project contains the API for the communication with the agrirouter. Everything you need for the
                     onboarding process, secure communication and much more.""",
     classifiers=[
-            'Intended Audience :: Developers',
-            'Natural Language :: English',
-            'License :: OSI Approved :: Apache Software License',
-            'Programming Language :: Python',
-            'Programming Language :: Python :: 3',
-            'Programming Language :: Python :: 3.6',
-            'Programming Language :: Python :: 3.7',
-            'Programming Language :: Python :: 3.8',
-            'Programming Language :: Python :: 3.9',
-            'Programming Language :: Python :: 3.10',
-        ],
+        'Intended Audience :: Developers',
+        'Natural Language :: English',
+        'License :: OSI Approved :: Apache Software License',
+        'Programming Language :: Python',
+        'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3.6',
+        'Programming Language :: Python :: 3.7',
+        'Programming Language :: Python :: 3.8',
+        'Programming Language :: Python :: 3.9',
+        'Programming Language :: Python :: 3.10',
+    ],
     install_requires=[
-            'certifi~=2021.5.30',
-            'cffi~=1.14.6',
-            'charset-normalizer~=2.0.6',
-            'cryptography~=3.4.8',
-            'idna~=3.2',
-            'pycparser~=2.20',
-            'requests~=2.26.0',
-            'urllib3~=1.26.7'
-        ],
+        'certifi~=2021.5.30',
+        'cffi~=1.14.6',
+        'charset-normalizer~=2.0.6',
+        'cryptography~=3.4.8',
+        'idna~=3.2',
+        'pycparser~=2.20',
+        'requests~=2.26.0',
+        'urllib3~=1.26.7'
+    ],
     project_urls={
         'Documentation': 'https://github.com/DKE-Data/agrirouter-sdk-python',
         'Source': 'https://github.com/DKE-Data/agrirouter-sdk-python',

tests/auth_test/test_auth.py

@@ -6,10 +6,8 @@
     public_key,
     private_key,
     auth_result_url,
-    ENV,
-    application_id,
+    ENV
 )
-import re
 
 
 class TestAuthorization:

tests/constants.py

@@ -121,7 +121,7 @@
 
 # taken from auth_result_url
 valid_response_token = (
-        "eyJhY2NvdW50IjoiMGJhMjRlZWUtYzMwYi00N2U1LWJkYzktNzcwM"
-        "2NmYjEzNmEwIiwicmVnY29kZSI6IjhlYWNiMTk4ZmMiLCJleHBpcm"
-        "VzIjoiMjAyMS0wOS0yM1QxNjowODo0My44ODhaIn0="
-    )
+    "eyJhY2NvdW50IjoiMGJhMjRlZWUtYzMwYi00N2U1LWJkYzktNzcwM"
+    "2NmYjEzNmEwIiwicmVnY29kZSI6IjhlYWNiMTk4ZmMiLCJleHBpcm"
+    "VzIjoiMjAyMS0wOS0yM1QxNjowODo0My44ODhaIn0="
+)

tests/enviroments_test/test_environments.py

@@ -1,157 +1,105 @@
 """Test agrirouter/environments/environments.py"""
 
-import pytest
-from agrirouter.environments.environments import ProductionEnvironment, QAEnvironment
-from tests.constants import application_id, auth_result_url
+from agrirouter.environments.environments import ProductionEnvironment as PE
+from agrirouter.environments.environments import QAEnvironment as QAE
+from tests.constants import application_id
 
 
-class TestProductionEnvironment:
+class TestPE:
     def test_get_base_url(self):
-        assert (
-            ProductionEnvironment().get_base_url()
-            == ProductionEnvironment._ENV_BASE_URL
-        )
+        assert PE().get_base_url() == PE._ENV_BASE_URL
 
     def test_get_api_prefix(self):
-        assert (
-            ProductionEnvironment().get_api_prefix()
-            == ProductionEnvironment._API_PREFIX
-        )
+        assert PE().get_api_prefix() == PE._API_PREFIX
 
     def test_get_registration_service_url(self):
-        assert (
-            ProductionEnvironment().get_registration_service_url()
-            == ProductionEnvironment._REGISTRATION_SERVICE_URL
-        )
+        assert PE().get_registration_service_url() == PE._REGISTRATION_SERVICE_URL
 
     def test_get_onboard_url(self):
-        assert (
-            ProductionEnvironment().get_onboard_url()
-            == ProductionEnvironment._REGISTRATION_SERVICE_URL
-            + ProductionEnvironment._API_PREFIX
-            + "/registration/onboard"
-        )
+        onb_url = PE._REGISTRATION_SERVICE_URL + PE._API_PREFIX + "/registration/onboard"
+        assert PE().get_onboard_url() == onb_url
 
     def test_get_secured_onboard_url(self):
-        assert (
-            ProductionEnvironment().get_secured_onboard_url()
-            == ProductionEnvironment._REGISTRATION_SERVICE_URL
-            + ProductionEnvironment._API_PREFIX
-            + "/registration/onboard/request"
-        )
+        onb_url = PE._REGISTRATION_SERVICE_URL + PE._API_PREFIX + "/registration/onboard/request"
+        assert PE().get_secured_onboard_url() == onb_url
 
     def test_get_verify_onboard_request_url(self):
-        assert (
-            ProductionEnvironment().get_verify_onboard_request_url()
-            == ProductionEnvironment._REGISTRATION_SERVICE_URL
-            + ProductionEnvironment._API_PREFIX
-            + "/registration/onboard/verify"
-        )
+        req_url = PE._REGISTRATION_SERVICE_URL + PE._API_PREFIX + "/registration/onboard/verify"
+        assert PE().get_verify_onboard_request_url() == req_url
 
     def test_get_revoke_url(self):
-        assert (
-            ProductionEnvironment().get_revoke_url()
-            == ProductionEnvironment._REGISTRATION_SERVICE_URL
-            + ProductionEnvironment._API_PREFIX
-            + "/registration/onboard/revoke"
-        )
+        rev_url = PE._REGISTRATION_SERVICE_URL + PE._API_PREFIX + "/registration/onboard/revoke"
+        assert PE().get_revoke_url() == rev_url
 
     def test_get_agrirouter_login_url(self):
-        assert (
-            ProductionEnvironment().get_agrirouter_login_url()
-            == ProductionEnvironment._ENV_BASE_URL
-            + ProductionEnvironment._AGRIROUTER_LOGIN_URL
-        )
+        login_url = PE._ENV_BASE_URL + PE._AGRIROUTER_LOGIN_URL
+        assert PE().get_agrirouter_login_url() == login_url
 
     def test_get_secured_onboarding_authorization_url(self):
         redirect_uri = "www.my_redirect.com"
         response_type = "response_type"
-        assert ProductionEnvironment().get_secured_onboarding_authorization_url(
+        assert PE().get_secured_onboarding_authorization_url(
             application_id, response_type, "state", redirect_uri
-        ) == "https://goto.my-agrirouter.com/application/{application_id}/authorize?response_type={response_type}&state={state}".format(
+        ) == "https://goto.my-agrirouter.com/application/{application_id}/authorize?response_type={response_type}&state={state}".format( # noqa
             application_id=application_id,
             response_type=response_type,
             state="state") + f"&redirect_uri={redirect_uri}"
 
     def test_get_mqtt_server_url(self):
-        assert ProductionEnvironment().get_mqtt_server_url(
+        assert PE().get_mqtt_server_url(
             "localhost", "5000"
-        ) == ProductionEnvironment._MQTT_URL_TEMPLATE.format(
+        ) == PE._MQTT_URL_TEMPLATE.format(
             host="localhost", port="5000"
         )
 
     def test_get_env_public_key(self):
-        assert (
-            ProductionEnvironment().get_env_public_key()
-            == ProductionEnvironment.AR_PUBLIC_KEY
-        )
+        assert PE().get_env_public_key() == PE.AR_PUBLIC_KEY
 
 
-class TestQAEnvironment:
+class TestQAE:
     def test_get_base_url(self):
-        assert QAEnvironment().get_base_url() == QAEnvironment._ENV_BASE_URL
+        assert QAE().get_base_url() == QAE._ENV_BASE_URL
 
     def test_get_api_prefix(self):
-        assert QAEnvironment().get_api_prefix() == QAEnvironment._API_PREFIX
+        assert QAE().get_api_prefix() == QAE._API_PREFIX
 
     def test_get_registration_service_url(self):
-        assert (
-            QAEnvironment().get_registration_service_url()
-            == QAEnvironment._REGISTRATION_SERVICE_URL
-        )
+        assert QAE().get_registration_service_url() == QAE._REGISTRATION_SERVICE_URL
 
     def test_get_onboard_url(self):
-        assert (
-            QAEnvironment().get_onboard_url()
-            == QAEnvironment._REGISTRATION_SERVICE_URL
-            + QAEnvironment._API_PREFIX
-            + "/registration/onboard"
-        )
+        onb_url = QAE._REGISTRATION_SERVICE_URL + QAE._API_PREFIX + "/registration/onboard"
+        assert QAE().get_onboard_url() == onb_url
 
     def test_get_secured_onboard_url(self):
-        assert (
-            QAEnvironment().get_secured_onboard_url()
-            == QAEnvironment._REGISTRATION_SERVICE_URL
-            + QAEnvironment._API_PREFIX
-            + "/registration/onboard/request"
-        )
+        onb_url = QAE._REGISTRATION_SERVICE_URL + QAE._API_PREFIX + "/registration/onboard/request"
+        assert QAE().get_secured_onboard_url() == onb_url
 
     def test_get_verify_onboard_request_url(self):
-        assert (
-            QAEnvironment().get_verify_onboard_request_url()
-            == QAEnvironment._REGISTRATION_SERVICE_URL
-            + QAEnvironment._API_PREFIX
-            + "/registration/onboard/verify"
-        )
+        req_url = QAE._REGISTRATION_SERVICE_URL + QAE._API_PREFIX + "/registration/onboard/verify"
+        assert QAE().get_verify_onboard_request_url() == req_url
 
     def test_get_revoke_url(self):
-        assert (
-            QAEnvironment().get_revoke_url()
-            == QAEnvironment._REGISTRATION_SERVICE_URL
-            + QAEnvironment._API_PREFIX
-            + "/registration/onboard/revoke"
-        )
+        rev_url = QAE._REGISTRATION_SERVICE_URL + QAE._API_PREFIX + "/registration/onboard/revoke"
+        assert QAE().get_revoke_url() == rev_url
 
     def test_get_agrirouter_login_url(self):
-        assert (
-            QAEnvironment().get_agrirouter_login_url()
-            == QAEnvironment._ENV_BASE_URL + QAEnvironment._AGRIROUTER_LOGIN_URL
-        )
+        login_url = QAE._ENV_BASE_URL + QAE._AGRIROUTER_LOGIN_URL
+        assert QAE().get_agrirouter_login_url() == login_url
 
     def test_get_secured_onboarding_authorization_url(self):
         redirect_uri = "www.my_redirect.com"
         response_type = "response_type"
-        assert QAEnvironment().get_secured_onboarding_authorization_url(
+        assert QAE().get_secured_onboarding_authorization_url(
             application_id, response_type, "state", redirect_uri
-        ) == QAEnvironment._ENV_BASE_URL + QAEnvironment._SECURED_ONBOARDING_AUTHORIZATION_LINK_TEMPLATE.format(
+        ) == QAE._ENV_BASE_URL + QAE._SECURED_ONBOARDING_AUTHORIZATION_LINK_TEMPLATE.format(
             application_id=application_id,
             response_type=response_type,
             state="state") + f"&redirect_uri={redirect_uri}"
 
     def test_get_mqtt_server_url(self):
-        assert QAEnvironment().get_mqtt_server_url(
+        assert QAE().get_mqtt_server_url(
             "localhost", "5000"
-        ) == QAEnvironment._MQTT_URL_TEMPLATE.format(host="localhost", port="5000")
+        ) == QAE._MQTT_URL_TEMPLATE.format(host="localhost", port="5000")
 
     def test_get_env_public_key(self):
-        assert QAEnvironment().get_env_public_key() == QAEnvironment.AR_PUBLIC_KEY
+        assert QAE().get_env_public_key() == QAE.AR_PUBLIC_KEY

tests/messaging_test/test_decode.py

@@ -1,12 +1,11 @@
 import json
-import pytest
 
 from agrirouter.generated.messaging.response.response_pb2 import ResponseEnvelope
 from agrirouter.messaging.decode import decode_response
 from agrirouter.messaging.decode import decode_details
 
 
-MESSAGING_RESULT = b'[{"sensorAlternateId":"185cd97b-ed0b-4e75-a6e2-6be1cdd38a06","capabilityAlternateId":"bbe9f361-b551-48d9-9fca-1b4dc768287c","command":{"message":"XwjIARAKGiQ5NWUzNWE0Zi1jNWM4LTQ1NDEtODE4OS03NmJlMzM0OTc0NDUiJDUzNzYyM2ZjLWY2NmYtNDc5Yi1hMmJhLWVjZjNlNWM3ZjhlMCoMCNTV5YsGEICI8LIDzQIKygIKTnR5cGVzLmFncmlyb3V0ZXIuY29tL2Fncmlyb3V0ZXIucmVzcG9uc2UucGF5bG9hZC5hY2NvdW50Lkxpc3RFbmRwb2ludHNSZXNwb25zZRL3AQp4CiRkNzA0YTQ0My05OWY3LTQ3YjQtYmU1NS1lMmZhMDk2ODllYmUSJFB5dGhvblNES19kZXYgLSAyMDIxLTEwLTI1LCAxMDo1MToxOBoLYXBwbGljYXRpb24iBmFjdGl2ZTIVdXJuOm15YXBwOnNucjAwMDAzMjM0CnsKJDE4NWNkOTdiLWVkMGItNGU3NS1hNmUyLTZiZTFjZGQzOGEwNhIkUHl0aG9uU0RLX2RldiAtIDIwMjEtMTAtMjEsIDIxOjQxOjI0GgthcHBsaWNhdGlvbiIGYWN0aXZlMhh1cm46bXlhcHA6c25yMDAwMDMyMzRzZGY="}}]'
+MESSAGING_RESULT = b'[{"sensorAlternateId":"185cd97b-ed0b-4e75-a6e2-6be1cdd38a06","capabilityAlternateId":"bbe9f361-b551-48d9-9fca-1b4dc768287c","command":{"message":"XwjIARAKGiQ5NWUzNWE0Zi1jNWM4LTQ1NDEtODE4OS03NmJlMzM0OTc0NDUiJDUzNzYyM2ZjLWY2NmYtNDc5Yi1hMmJhLWVjZjNlNWM3ZjhlMCoMCNTV5YsGEICI8LIDzQIKygIKTnR5cGVzLmFncmlyb3V0ZXIuY29tL2Fncmlyb3V0ZXIucmVzcG9uc2UucGF5bG9hZC5hY2NvdW50Lkxpc3RFbmRwb2ludHNSZXNwb25zZRL3AQp4CiRkNzA0YTQ0My05OWY3LTQ3YjQtYmU1NS1lMmZhMDk2ODllYmUSJFB5dGhvblNES19kZXYgLSAyMDIxLTEwLTI1LCAxMDo1MToxOBoLYXBwbGljYXRpb24iBmFjdGl2ZTIVdXJuOm15YXBwOnNucjAwMDAzMjM0CnsKJDE4NWNkOTdiLWVkMGItNGU3NS1hNmUyLTZiZTFjZGQzOGEwNhIkUHl0aG9uU0RLX2RldiAtIDIwMjEtMTAtMjEsIDIxOjQxOjI0GgthcHBsaWNhdGlvbiIGYWN0aXZlMhh1cm46bXlhcHA6c25yMDAwMDMyMzRzZGY="}}]' # noqa
 
 
 def test_decode_response():
@@ -17,8 +16,8 @@ def test_decode_response():
 
     assert message.response_payload.details
 
-    assert message.response_payload.details.type_url == "types.agrirouter.com/agrirouter.response.payload.account.ListEndpointsResponse"
-    assert message.response_payload.details.value == b'\nx\n$d704a443-99f7-47b4-be55-e2fa09689ebe\x12$PythonSDK_dev - 2021-10-25, 10:51:18\x1a\x0bapplication"\x06active2\x15urn:myapp:snr00003234\n{\n$185cd97b-ed0b-4e75-a6e2-6be1cdd38a06\x12$PythonSDK_dev - 2021-10-21, 21:41:24\x1a\x0bapplication"\x06active2\x18urn:myapp:snr00003234sdf'
+    assert message.response_payload.details.type_url == "types.agrirouter.com/agrirouter.response.payload.account.ListEndpointsResponse" # noqa
+    assert message.response_payload.details.value == b'\nx\n$d704a443-99f7-47b4-be55-e2fa09689ebe\x12$PythonSDK_dev - 2021-10-25, 10:51:18\x1a\x0bapplication"\x06active2\x15urn:myapp:snr00003234\n{\n$185cd97b-ed0b-4e75-a6e2-6be1cdd38a06\x12$PythonSDK_dev - 2021-10-21, 21:41:24\x1a\x0bapplication"\x06active2\x18urn:myapp:snr00003234sdf' # noqa
 
     assert message.response_envelope.response_code == 200
     assert message.response_envelope.type == ResponseEnvelope.ResponseBodyType.Value("ENDPOINTS_LISTING")

tests/onboarding_test/test_headers.py

@@ -15,8 +15,7 @@ def test_get_header(self):
             self.test_object.get_header()["Authorization"] == "Bearer " + self.reg_code
         )
         assert (
-            self.test_object.get_header()["Content-Type"]
-            == ContentTypes.APPLICATION_JSON.value
+            self.test_object.get_header()["Content-Type"] == ContentTypes.APPLICATION_JSON.value
         )
 
         assert self.test_object_1.get_header()["Content-Type"] == "json"

tests/onboarding_test/test_request_onboarding.py

@@ -46,6 +46,5 @@ def test_get_header(self):
         )
         assert self.test_object.get_header()["Content-Type"] == self.content_type
         assert (
-            self.test_object.get_header()["X-Agrirouter-ApplicationId"]
-            == application_id
+            self.test_object.get_header()["X-Agrirouter-ApplicationId"] == application_id
         )

tests/onboarding_test/test_signature.py

@@ -1,12 +1,11 @@
 """Test agrirouter/onboarding/signature.py"""
 
 import pytest
-import re
 
 from cryptography.exceptions import InvalidSignature
 
 from agrirouter.onboarding.signature import create_signature, verify_signature
-from tests.constants import private_key, wrong_private_key, public_key, valid_response_signature
+from tests.constants import private_key, public_key
 
 
 def test_create_signature_ok():

tox.ini

@@ -4,11 +4,13 @@ exclude =
     .git,
     __pycache__,
     .pytest_cache
+    ./agrirouter/__init__.py,
     agrirouter_sdk_python.egg-info,
     assets,
     old,
     build,
     dist,
+    example_script.py,
     venv,
     conftest.py,
     *_pb2.py