Improve Python cosmetics

tests/integration/conftest.py

@@ -1,69 +1,71 @@
 """Wakaama integration tests (pytest) fixtures"""
 from pathlib import Path
 from time import sleep
-import pytest
+
 import pexpect
+import pytest
 
 
 # pylint: disable=no-member
 class HelperBase:
     """Provides helper methods for integration tests. Wraps "pexpect" API"""
 
-    def __dumptext(self):
+    def __dump_text(self):
         """Internal method that prints debug help"""
         print("Debug help: actual output---------------------")
         print(self.pexpectobj.before)
         print("----------------------------------------------")
 
-    def commandresponse(self, cmd,resp):
+    def command_response(self, cmd, resp):
         """Issue CLI command and check response"""
         self.pexpectobj.sendline(cmd)
         try:
             self.pexpectobj.expect_exact(resp)
         except pexpect.exceptions.TIMEOUT:
-            self.__dumptext()
+            self.__dump_text()
             return False
         return True
 
-    def waitforpacket(self):
+    def wait_for_packet(self):
         """Wait for "packet printout" and return text output"""
         try:
             self.pexpectobj.expect_exact("bytes received from")
         except pexpect.exceptions.TIMEOUT:
-            self.__dumptext()
+            self.__dump_text()
             assert False
         try:
             self.pexpectobj.expect_exact("\r\r\n>")
         except pexpect.exceptions.TIMEOUT:
-            self.__dumptext()
+            self.__dump_text()
             assert False
         return self.pexpectobj.before
 
-    def waitfortext(self,txt):
+    def wait_for_text(self, txt):
         """Wait for text printout and return True if found"""
         try:
             self.pexpectobj.expect_exact(txt)
         except pexpect.exceptions.TIMEOUT:
-            self.__dumptext()
+            self.__dump_text()
             return False
         return True
 
-    def waitfortime(self, timedelay):
+    def wait_for_time(self, timedelay):
         """Wait for time and return output since last command"""
         sleep(timedelay)
         # this is a "hack" to be able to return output between commands
-        self.commandresponse("help", "help")
+        self.command_response("help", "help")
         return self.pexpectobj.before
 
     def quit(self):
         """Quit client or server"""
-        self.pexpectobj.sendline("q")       # exit, if client
-        self.pexpectobj.sendline("quit")    # exit, if server
-        self.pexpectobj.expect(pexpect.EOF) # make sure exited
+        self.pexpectobj.sendline("q")  # exit, if client
+        self.pexpectobj.sendline("quit")  # exit, if server
+        self.pexpectobj.expect(pexpect.EOF)  # make sure exited
 
 
 class Lwm2mServer(HelperBase):
     """Server subclass of HelperBase"""
+
     def __init__(self, arguments="", timeout=3, encoding="utf8"):
         base_path = str(Path(__file__).parent.absolute())
         self.pexpectobj = pexpect.spawn(base_path +
@@ -77,8 +79,10 @@ def __init__(self, arguments="", timeout=3, encoding="utf8"):
                                        "w",
                                        encoding="utf-8")
 
+
 class Lwm2mClient(HelperBase):
     """Client subclass of HelperBase"""
+
     def __init__(self, arguments="", timeout=3, encoding="utf8"):
         base_path = str(Path(__file__).parent.absolute())
         self.pexpectobj = pexpect.spawn(base_path +
@@ -92,8 +96,10 @@ def __init__(self, arguments="", timeout=3, encoding="utf8"):
                                        "w",
                                        encoding="utf-8")
 
+
 class Lwm2mBootstrapServer(HelperBase):
     """Bootstrap-server subclass of HelperBase"""
+
     def __init__(self, arguments="", timeout=3, encoding="utf8"):
         base_path = str(Path(__file__).parent.absolute())
         self.pexpectobj = pexpect.spawn(base_path +
@@ -137,9 +143,9 @@ def lwm2mclient(request):
 
 @pytest.fixture
 def lwm2mbootstrapserver():
-    """Provide lwm2mclient instance."""
+    """Provide bootstrap_server instance."""
     bootstrapserver = Lwm2mBootstrapServer("-f examples/bootstrap_server/bootstrap_server.ini",
                                            timeout=13)
-    bootstrapserver.waitfortext(">")
+    bootstrapserver.wait_for_text(">")
     yield bootstrapserver
     bootstrapserver.quit()

tests/integration/test_bootstrap_interface.py

@@ -10,48 +10,48 @@ def test_client_initiated_bootstrap(lwm2mbootstrapserver, lwm2mclient):
     Client Initiated Bootstrap Mode"""
 
     # check that client attempts to bootstrap
-    assert lwm2mclient.waitfortext("STATE_BOOTSTRAPPING")
+    assert lwm2mclient.wait_for_text("STATE_BOOTSTRAPPING")
 
     # Test Procedure 1
     # bootstrap request is triggered by command line option
-    assert lwm2mbootstrapserver.waitfortext('Bootstrap request from "apa"\r\r\n')
+    assert lwm2mbootstrapserver.wait_for_text('Bootstrap request from "apa"\r\r\n')
 
     # Pass-Criteria A
-    # should check for "2.04" changed but informarmation not available at CLI.
+    # should check for "2.04" changed but information not available at CLI.
     # Just check that we got packet (of correct size)
-    assert lwm2mclient.waitfortext("8 bytes received from")
+    assert lwm2mclient.wait_for_text("8 bytes received from")
 
     # check that DELETE /0 is sent and ack:ed (not in TC spec)
-    assert lwm2mbootstrapserver.waitfortext('Sending DELETE /0 to "apa" OK.')
-    assert lwm2mbootstrapserver.waitfortext("Received status 2.02 (COAP_202_DELETED)"
-                                            " for URI /0 from endpoint apa.")
+    assert lwm2mbootstrapserver.wait_for_text('Sending DELETE /0 to "apa" OK.')
+    assert lwm2mbootstrapserver.wait_for_text("Received status 2.02 (COAP_202_DELETED)"
+                                              " for URI /0 from endpoint apa.")
 
     # check that DELETE /1 is sent and ack:ed (not in TC spec)
-    assert lwm2mbootstrapserver.waitfortext('Sending DELETE /1 to "apa" OK.')
-    assert lwm2mbootstrapserver.waitfortext("Received status 2.02 (COAP_202_DELETED)"
-                                            " for URI /1 from endpoint apa.")
+    assert lwm2mbootstrapserver.wait_for_text('Sending DELETE /1 to "apa" OK.')
+    assert lwm2mbootstrapserver.wait_for_text("Received status 2.02 (COAP_202_DELETED)"
+                                              " for URI /1 from endpoint apa.")
 
     # check that bootstrap write operations are sent (not checking data)
-    assert lwm2mbootstrapserver.waitfortext('Sending WRITE /0/1 to "apa" OK.')
+    assert lwm2mbootstrapserver.wait_for_text('Sending WRITE /0/1 to "apa" OK.')
     # Pass-Criteria C
     # check that bootstrap server received success ACK
-    assert lwm2mbootstrapserver.waitfortext("Received status 2.04 (COAP_204_CHANGED)"
-                                            " for URI /0/1 from endpoint apa.")
+    assert lwm2mbootstrapserver.wait_for_text("Received status 2.04 (COAP_204_CHANGED)"
+                                              " for URI /0/1 from endpoint apa.")
 
     # check that bootstrap write operations are sent (not checking data)
-    assert lwm2mbootstrapserver.waitfortext('Sending WRITE /1/1 to "apa" OK.')
+    assert lwm2mbootstrapserver.wait_for_text('Sending WRITE /1/1 to "apa" OK.')
     # Pass-Criteria C
     # check that bootstrap server received success ACK
-    assert lwm2mbootstrapserver.waitfortext("Received status 2.04 (COAP_204_CHANGED)"
-                                            " for URI /1/1 from endpoint apa.")
+    assert lwm2mbootstrapserver.wait_for_text("Received status 2.04 (COAP_204_CHANGED)"
+                                              " for URI /1/1 from endpoint apa.")
 
     # Pass-Criteria D
     # check bootstrap discover ACK
-    assert lwm2mbootstrapserver.waitfortext("Received status 2.05 (COAP_205_CONTENT)"
-                                            " for URI / from endpoint apa.")
+    assert lwm2mbootstrapserver.wait_for_text("Received status 2.05 (COAP_205_CONTENT)"
+                                              " for URI / from endpoint apa.")
 
     # Pass-Criteria E
     # check that bootstrap server receives a successful ACK for bootstrap finish
-    assert lwm2mbootstrapserver.waitfortext('Sending BOOTSTRAP FINISH  to "apa" OK.')
-    assert lwm2mbootstrapserver.waitfortext("Received status 2.04 (COAP_204_CHANGED)"
-                                            " for URI / from endpoint apa.")
+    assert lwm2mbootstrapserver.wait_for_text('Sending BOOTSTRAP FINISH  to "apa" OK.')
+    assert lwm2mbootstrapserver.wait_for_text("Received status 2.04 (COAP_204_CHANGED)"
+                                              " for URI / from endpoint apa.")

tests/integration/test_device_mgmnt_interface.py

@@ -1,6 +1,7 @@
 """Wakaama integration tests (pytest)"""
-import re
 import json
+import re
+
 from helpers.helpers import get_senml_json_record
 
 
@@ -9,10 +10,10 @@ def test_read_on_object(lwm2mserver, lwm2mclient):
     Purpose of this test is to show conformance with the LwM2M Read
     operation on whole Object"""
 
-    lwm2mclient.waitfortext("STATE_READY")
+    lwm2mclient.wait_for_text("STATE_READY")
     # Test Procedure 1
-    assert lwm2mserver.commandresponse("read 0 /1", "OK")
-    text = lwm2mserver.waitforpacket()
+    assert lwm2mserver.command_response("read 0 /1", "OK")
+    text = lwm2mserver.wait_for_packet()
     # Pass-Criteria A
     assert text.find("COAP_205_CONTENT") > 0
     packet = re.findall(r"(\[.*\])", text)
@@ -26,8 +27,8 @@ def test_read_on_object(lwm2mserver, lwm2mclient):
     assert get_senml_json_record(parsed, "0/6", "vb") is False
     assert get_senml_json_record(parsed, "0/7", "vs") == 'U'
     # Test Procedure 2
-    assert lwm2mserver.commandresponse("read 0 /3", "OK")
-    text = lwm2mserver.waitforpacket()
+    assert lwm2mserver.command_response("read 0 /3", "OK")
+    text = lwm2mserver.wait_for_packet()
     # Pass-Criteria B
     assert text.find("COAP_205_CONTENT") > 0
     packet = re.findall(r"(\[.*\])", text)
@@ -48,7 +49,7 @@ def test_read_on_object(lwm2mserver, lwm2mclient):
     assert get_senml_json_record(parsed, "0/9", "v") == 100
     assert get_senml_json_record(parsed, "0/10", "v") == 15
     assert get_senml_json_record(parsed, "0/11/0", "v") == 0
-    assert get_senml_json_record(parsed, "0/13", "v") > 0    # current time
+    assert get_senml_json_record(parsed, "0/13", "v") > 0  # current time
     assert get_senml_json_record(parsed, "0/14", "vs") == "+01:00"
     assert get_senml_json_record(parsed, "0/15", "vs") == "Europe/Berlin"
     assert get_senml_json_record(parsed, "0/16", "vs") == "U"

tests/integration/test_information_reporting_interface.py

@@ -1,7 +1,9 @@
 """Wakaama integration tests (pytest)"""
-import re
 import json
+import re
+
 import pytest
+
 from helpers.helpers import get_senml_json_record
 
 
@@ -13,21 +15,21 @@ def test_observation_notification(lwm2mserver, lwm2mclient):
     Sending the observation policy to the device."""
 
     # Test Procedure 1
-    assert lwm2mserver.commandresponse("read 0 /3/0/6", "OK")
-    text = lwm2mserver.waitforpacket()
+    assert lwm2mserver.command_response("read 0 /3/0/6", "OK")
+    text = lwm2mserver.wait_for_packet()
     assert text.find("COAP_205_CONTENT") > 0
     packet = re.findall(r"(\[.*\])", text)
     parsed = json.loads(packet[1])
-    assert next(item for item in parsed if item["n"] == "0")["v"] == 1 # Internal battery
-    assert next(item for item in parsed if item["n"] == "1")["v"] == 5 # USB
+    assert next(item for item in parsed if item["n"] == "0")["v"] == 1  # Internal battery
+    assert next(item for item in parsed if item["n"] == "1")["v"] == 5  # USB
     # Test Procedure 2
-    assert lwm2mserver.commandresponse("time 0 /3/0/7 5 15", "OK")
-    assert lwm2mserver.commandresponse("time 0 /3/0/8 10 20", "OK")
+    assert lwm2mserver.command_response("time 0 /3/0/7 5 15", "OK")
+    assert lwm2mserver.command_response("time 0 /3/0/8 10 20", "OK")
     # Test Procedure 3
-    assert lwm2mserver.commandresponse("observe 0 /3/0/7", "OK")
-    assert lwm2mserver.commandresponse("observe 0 /3/0/8", "OK") #consumes first observe from /3/0/7
-    text = lwm2mserver.waitforpacket() # consumes ACK and piggy-backed first observe from /3/0/8
-    text = lwm2mserver.waitfortime(20) # wait for max "pmax"
+    assert lwm2mserver.command_response("observe 0 /3/0/7", "OK")
+    assert lwm2mserver.command_response("observe 0 /3/0/8", "OK")  # consume 1st observe from /3/0/7
+    text = lwm2mserver.wait_for_packet()  # consumes ACK and piggy-backed first observe from /3/0/8
+    text = lwm2mserver.wait_for_time(20)  # wait for max "pmax"
     # Pass-Criteria A
     assert 0 < text.count("Notify from client #0 /3/0/7") <= 4
     assert 0 < text.count("Notify from client #0 /3/0/8") <= 2
@@ -51,24 +53,24 @@ def test_cancel_observations_using_reset(lwm2mserver, lwm2mclient):
     Cancel the Observation relationship by sending “Cancel Observation”
     operation."""
     # Test Procedure 1
-    assert lwm2mserver.commandresponse("time 0 /3/0/7 3 6", "OK") # faster timeout vs int-301
-    assert lwm2mserver.commandresponse("time 0 /3/0/8 6 7", "OK")
-    assert lwm2mserver.commandresponse("observe 0 /3/0/7", "OK")
-    assert lwm2mserver.commandresponse("observe 0 /3/0/8", "OK")
-    text = lwm2mserver.waitfortime(15)
+    assert lwm2mserver.command_response("time 0 /3/0/7 3 6", "OK")  # faster timeout vs int-301
+    assert lwm2mserver.command_response("time 0 /3/0/8 6 7", "OK")
+    assert lwm2mserver.command_response("observe 0 /3/0/7", "OK")
+    assert lwm2mserver.command_response("observe 0 /3/0/8", "OK")
+    text = lwm2mserver.wait_for_time(15)
     # Pass-Criteria A
     # Make sure that notifications are sent
     assert text.count("Notify from client #0 /3/0/7") >= 2
     assert text.count("Notify from client #0 /3/0/8") >= 2
     # Test Procedure 2
-    assert lwm2mserver.commandresponse("cancel 0 /3/0/7", "OK")
-    text = lwm2mserver.waitfortime(15)
+    assert lwm2mserver.command_response("cancel 0 /3/0/7", "OK")
+    text = lwm2mserver.wait_for_time(15)
     # Pass-Criteria B
     assert text.count("Notify from client #0 /3/0/7") == 0
     assert text.count("Notify from client #0 /3/0/8") >= 2
     # Test Procedure 3, 4
-    assert lwm2mserver.commandresponse("cancel 0 /3/0/8", "OK")
-    text = lwm2mserver.waitfortime(15)
+    assert lwm2mserver.command_response("cancel 0 /3/0/8", "OK")
+    text = lwm2mserver.wait_for_time(15)
     # Pass-Criteria B cont.
     assert text.count("Notify from client #0 /3/0/7") == 0
     assert text.count("Notify from client #0 /3/0/8") == 0

tests/integration/test_registration_interface.py

@@ -1,5 +1,6 @@
 """Wakaama integration tests (pytest)"""
 import re
+
 import pytest
 
 
@@ -31,8 +32,8 @@ def test_registration_interface(lwm2mserver, lwm2mclient):
     """LightweightM2M-1.1-int-101
     Test that the Client registers with the Server."""
 
-    lwm2mclient.waitfortext("STATE_READY")
-    text = lwm2mserver.waitforpacket()
+    lwm2mclient.wait_for_text("STATE_READY")
+    text = lwm2mserver.wait_for_packet()
     client_id, event, endpoint, version, binding, lifetime, objects = \
         parse_client_registration(text)
     assert client_id == 0
@@ -50,16 +51,16 @@ def test_registration_information_update(lwm2mserver, lwm2mclient):
     """LightweightM2M-1.1-int-102
     Test that the client updates the registration information on the Server."""
 
-    lwm2mclient.waitfortext("STATE_READY")
+    lwm2mclient.wait_for_text("STATE_READY")
     # Test Procedure 1
-    assert lwm2mserver.commandresponse("write 0 /1/0/1 20", "OK")
-    text = lwm2mserver.waitforpacket()
+    assert lwm2mserver.command_response("write 0 /1/0/1 20", "OK")
+    text = lwm2mserver.wait_for_packet()
     # Pass-Criteria A
     assert text.find("COAP_204_CHANGED") > 0
     # Test Procedure 2
-    text = lwm2mserver.waitfortime(20)
+    text = lwm2mserver.wait_for_time(20)
     # Pass-Criteria B, D (C not observable)
-    assert text.find("lifetime: 300 sec") > 0 # NOTE: this should be 20 sec?
+    assert text.find("lifetime: 300 sec") > 0  # NOTE: this should be 20 sec?
     # expect update response immediately and again after every half "lifetime" (10s)
     assert 2 <= text.count("Client #0 updated.") <= 3
 
@@ -74,22 +75,22 @@ def test_discarded_register_update(lwm2mserver, lwm2mclient):
 
     Test only partially implemented due to limitations (see comments)"""
 
-    lwm2mclient.waitfortext("STATE_READY")
+    lwm2mclient.wait_for_text("STATE_READY")
     # Test Procedure 1
     # server learns client lifetime when client registers
     # a server write to /1/x/1 does not affect the servers knowledge about
     # client lifetime therefore a higher value than the inital registration lifetime
     # is written to this resource to provoke the server to drop the client
-    text = lwm2mserver.waitfortime(60)
+    text = lwm2mserver.wait_for_time(60)
     # Pass-Criteria A
     assert text.find("Client #0 updated.")
     # Test Procedure 2
     # Client reports 60s lifetime to server
     # Server changes client lifetime to 180
     #   => server will drop client before an update is received
-    assert lwm2mserver.commandresponse("write 0 /1/0/1 180", "OK")
+    assert lwm2mserver.command_response("write 0 /1/0/1 180", "OK")
     # ensure client is dropped >60s (in practice >30s)
-    text = lwm2mserver.waitfortime(70)
+    text = lwm2mserver.wait_for_time(70)
     # Pass-Criteria B
     assert text.find("Client #0 unregistered.") > 0
     # Test Procedure 3