Poll until code

Author: razor-x

seamapi/access_codes.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone, timedelta
 from seamapi.types import (
     AbstractAccessCodes,
     AccessCode,
@@ -166,22 +167,27 @@ def create(
         if common_code_key is not None:
             create_payload["common_code_key"] = common_code_key
 
+        if (wait_for_code
+            and starts_at is not None
+            and datetime.fromisoformat(starts_at) > datetime.now() + timedelta(seconds=5)
+        ):
+            raise RuntimeException("Cannot use wait_for_code with a future time bound code")
+
         res = self.seam.make_request(
             "POST",
             "/access_codes/create",
             json=create_payload,
         )
 
-        # TODO: Poll AccessCode not ActionAttempt
-        if wait_for_code:
-            action_attempt = self.seam.action_attempts.poll_until_ready(
-                res["action_attempt"]["action_attempt_id"]
-            )
-            success_res: Any = action_attempt.result
+        access_code = AccessCode.from_dict(res["access_code"])
 
-            return AccessCode.from_dict(success_res["access_code"])
+        if wait_for_code:
+            while (access_code.code is None):
+                if (access_code.status != "unknown"):
+                    raise RuntimeException("Gave up waiting for code since access code status is unknown")
+                access_code = access_codes.get(access_code)
 
-        return AccessCode.from_dict(res["access_code"])
+        return access_code
 
     @report_error
     def create_multiple(

tests/access_codes/test_access_codes.py

@@ -31,13 +31,20 @@ def test_access_codes(seam: Seam):
     delete_action_attempt = seam.access_codes.delete(created_access_code)
     assert delete_action_attempt.status == "success"
 
-    created_access_code = seam.access_codes.create(
-        some_device.device_id, "Other Test code", "4445", wait_for_code=False
-    )
-    assert created_access_code.name == "Other Test code"
-    assert created_access_code.code == "4445"
-
     # TODO: Can only test for salto devices.
     # access_codes = seam.access_codes.create_multiple(devices=all_devices)
     # assert len(access_codes) == len(all_devices)
     # assert len(set([ac.common_code_key for ac in access_codes])) == 1
+
+def test_access_codes_create_wait_for_code(seam: Seam):
+    run_august_factory(seam)
+
+    all_devices = seam.devices.list()
+    some_device = all_devices[0]
+
+    created_access_code = seam.access_codes.create(
+        some_device.device_id, "Test code", "4445", wait_for_code=True
+    )
+
+    assert created_access_code.name == "Test code"
+    assert created_access_code.code == "4445"