Porting tests from archived repository of OpenPAYGO-Token

.gitignore

@@ -1,4 +1,6 @@
 *.pyc
 dist
 openpaygo.egg-info
-.DS_store
+.DS_store
+__pycache__
+venv

openpaygo/metrics_shared.py

@@ -146,10 +146,27 @@ def generate_hash_string(cls, input_string, secret_key):
 
     @classmethod
     def load_secret_key_from_hex(cls, secret_key):
+        if isinstance(secret_key, (bytes, bytearray)):
+            secret_key_bytes = bytes(secret_key)
+            if len(secret_key_bytes) != 16:
+                raise ValueError(
+                    "The secret key provided is not correctly formatted, it should be 16 "
+                    "bytes. "
+                )
+            return secret_key_bytes
+
         try:
-            return codecs.decode(secret_key, "hex")
+            decoded = codecs.decode(secret_key, "hex")
         except Exception:
             raise ValueError(
                 "The secret key provided is not correctly formatted, it should be 32 "
                 "hexadecimal characters. "
             )
+
+        if len(decoded) != 16:
+            raise ValueError(
+                "The secret key provided is not correctly formatted, it should be 32 "
+                "hexadecimal characters. "
+            )
+
+        return decoded

openpaygo/simulators/__init__.py

@@ -0,0 +1,4 @@
+from .device_simulator import DeviceSimulator as DeviceSimulator
+from .server_simulator import SingleDeviceServerSimulator as SingleDeviceServerSimulator
+
+__all__ = ["DeviceSimulator", "SingleDeviceServerSimulator"]

openpaygo/simulators/device_simulator.py

@@ -0,0 +1,159 @@
+from datetime import datetime, timedelta
+
+from openpaygo.token_decode import OpenPAYGOTokenDecoder, TokenType
+from openpaygo.token_shared import OpenPAYGOTokenShared
+
+
+class DeviceSimulator(object):
+
+    def __init__(
+        self,
+        starting_code,
+        key,
+        starting_count=1,
+        restricted_digit_set=False,
+        waiting_period_enabled=True,
+        time_divider=1,
+    ):
+        self.starting_code = starting_code
+        self.key = key
+        self.time_divider = time_divider
+        self.restricted_digit_set = restricted_digit_set
+        self.waiting_period_enabled = (
+            waiting_period_enabled  # Should always be true except for testing
+        )
+
+        self.payg_enabled = True
+        self.count = starting_count
+        self.expiration_date = datetime.now()
+        self.invalid_token_count = 0
+        self.token_entry_blocked_until = datetime.now()
+        self.used_counts = []
+
+    def print_status(self):
+        print('-------------------------')
+        print('Expiration Date: '+ str(self.expiration_date))
+        print('Current count: '+str(self.count))
+        print('PAYG Enabled: '+str(self.payg_enabled))
+        print('Active: '+str(self.is_active()))
+        print('-------------------------')
+
+    def is_active(self):
+        return self.expiration_date > datetime.now()
+
+    def enter_token(self, token, show_result=True):
+        if len(token) == 9:
+            token_int = int(token)
+            return self._update_device_status_from_token(token_int, show_result)
+        else:
+            token_int = int(token)
+            return self._update_device_status_from_extended_token(token_int, show_result)
+
+    def get_days_remaining(self):
+        if self.payg_enabled:
+            td = self.expiration_date - datetime.now()
+            days, hours, minutes = td.days, td.seconds//3600, (td.seconds//60)%60
+            days = days + (hours + minutes/60)/24
+            return round(days)
+        else:
+            return 'infinite'
+
+    def _update_device_status_from_token(self, token, show_result=True):
+        if (
+            self.token_entry_blocked_until > datetime.now()
+            and self.waiting_period_enabled
+        ):
+            if show_result:
+                print('TOKEN_ENTRY_BLOCKED')
+            return False
+
+        token_value, token_type, token_count, updated_counts = (
+            OpenPAYGOTokenDecoder.get_activation_value_count_and_type_from_token(
+                token=token,
+                starting_code=self.starting_code,
+                key=self.key,
+                last_count=self.count,
+                restricted_digit_set=self.restricted_digit_set,
+                used_counts=self.used_counts,
+            )
+        )
+        if token_value is None:
+            if token_type == TokenType.ALREADY_USED:
+                if show_result:
+                    print('OLD_TOKEN')
+                return -2
+            if show_result:
+                print('TOKEN_INVALID')
+            self.invalid_token_count += 1
+            self.token_entry_blocked_until = datetime.now() + timedelta(
+                minutes=2**self.invalid_token_count
+            )
+            return -1
+        elif token_value == -2:
+            if show_result:
+                print('OLD_TOKEN')
+            return -2
+        else:
+            if show_result:
+                print('TOKEN_VALID', ' | Value:', token_value)
+            if (
+                token_count > self.count
+                or token_value == OpenPAYGOTokenShared.COUNTER_SYNC_VALUE
+            ):
+                self.count = token_count
+            self.used_counts = updated_counts
+            self.invalid_token_count = 0
+            self._update_device_status_from_token_value(token_value, token_type)
+            return 1
+
+    def _update_device_status_from_extended_token(self, token, show_result=True):
+        if (
+            self.token_entry_blocked_until > datetime.now()
+            and self.waiting_period_enabled
+        ):
+            if show_result:
+                print('TOKEN_ENTRY_BLOCKED')
+
+        token_value, token_count = (
+            OpenPAYGOTokenDecoder.get_activation_value_count_from_extended_token(
+                token=token,
+                starting_code=self.starting_code,
+                key=self.key,
+                last_count=self.count,
+                restricted_digit_set=self.restricted_digit_set,
+            )
+        )
+        if token_value is None:
+            if show_result:
+                print('TOKEN_INVALID')
+            self.invalid_token_count += 1
+            self.token_entry_blocked_until = datetime.now() + timedelta(
+                minutes=2**self.invalid_token_count
+            )
+        else:
+            if show_result:
+                print('Special token entered, value: '+str(token_value))
+
+    def _update_device_status_from_token_value(self, token_value, token_type):
+        if token_value <= OpenPAYGOTokenShared.MAX_ACTIVATION_VALUE:
+            if not self.payg_enabled and token_type == TokenType.SET_TIME:
+                self.payg_enabled = True
+            if self.payg_enabled:
+                self._update_expiration_date_from_value(token_value, token_type)
+        elif token_value == OpenPAYGOTokenShared.PAYG_DISABLE_VALUE:
+            self.payg_enabled = False
+        elif token_value != OpenPAYGOTokenShared.COUNTER_SYNC_VALUE:
+            # We do nothing if its the sync counter value, the counter has been synced already
+            print('COUNTER_SYNCED')
+        else:
+            # If it's another value we also do nothing, as they are not defined
+            print('UNKNOWN_COMMAND')
+
+    def _update_expiration_date_from_value(self, toke_value, token_type):
+        number_of_days = toke_value/self.time_divider
+        if token_type == TokenType.SET_TIME:
+            self.expiration_date = datetime.now() + timedelta(days=number_of_days)
+        else:
+            if self.expiration_date < datetime.now():
+                self.expiration_date = datetime.now()
+            self.expiration_date = self.expiration_date + timedelta(days=number_of_days)

openpaygo/simulators/server_simulator.py

@@ -0,0 +1,110 @@
+from datetime import datetime
+
+from openpaygo.token_encode import OpenPAYGOTokenEncoder
+from openpaygo.token_shared import OpenPAYGOTokenShared, TokenType
+
+
+class SingleDeviceServerSimulator(object):
+
+    def __init__(
+        self,
+        starting_code,
+        key,
+        starting_count=1,
+        restricted_digit_set=False,
+        time_divider=1,
+    ):
+        self.starting_code = starting_code
+        self.key = key
+        self.count = starting_count
+        self.expiration_date = datetime.now()
+        self.furthest_expiration_date = datetime.now()
+        self.payg_enabled = True
+        self.time_divider = time_divider
+        self.restricted_digit_set = restricted_digit_set
+
+    def print_status(self):
+        print('Expiration Date: '+ str(self.expiration_date))
+        print('Current count: '+str(self.count))
+        print('PAYG Enabled: '+str(self.payg_enabled))
+
+    def generate_payg_disable_token(self):
+        self.count, token = OpenPAYGOTokenEncoder.generate_token(
+            starting_code=self.starting_code,
+            secret_key=self.key,
+            value=0,
+            count=self.count,
+            token_type=TokenType.DISABLE_PAYG,
+            restricted_digit_set=self.restricted_digit_set
+        )
+        return SingleDeviceServerSimulator._format_token(token)
+
+    def generate_counter_sync_token(self):
+        self.count, token = OpenPAYGOTokenEncoder.generate_token(
+            starting_code=self.starting_code,
+            secret_key=self.key,
+            value=0,
+            count=self.count,
+            token_type=TokenType.COUNTER_SYNC,
+            restricted_digit_set=self.restricted_digit_set
+        )
+        return SingleDeviceServerSimulator._format_token(token)
+
+    def generate_token_from_date(self, new_expiration_date, force=False):
+        furthest_expiration_date = self.furthest_expiration_date
+        if new_expiration_date > self.furthest_expiration_date:
+            self.furthest_expiration_date = new_expiration_date
+
+        if new_expiration_date > furthest_expiration_date:
+            # If the date is strictly above the furthest date activated, use ADD
+            value = self._get_value_to_activate(
+                new_expiration_date, self.expiration_date, force
+            )
+            self.expiration_date = new_expiration_date
+            return self._generate_token_from_value(value, mode=TokenType.ADD_TIME)
+        else:
+            # If the date is below or equal to the furthest date activated, use SET
+            value = self._get_value_to_activate(
+                new_expiration_date, datetime.now(), force
+            )
+            self.expiration_date = new_expiration_date
+            return self._generate_token_from_value(value, mode=TokenType.SET_TIME)
+
+    def _generate_token_from_value(self, value, mode):
+        self.count, token = OpenPAYGOTokenEncoder.generate_token(
+            starting_code=self.starting_code,
+            secret_key=self.key,
+            value=value,
+            count=self.count,
+            token_type=mode,
+            restricted_digit_set=self.restricted_digit_set
+        )
+        return SingleDeviceServerSimulator._format_token(token)
+
+    def _generate_extended_value_token(self, value):
+        pass
+
+    def _get_value_to_activate(self, new_time, reference_time, force_maximum=False):
+        if new_time <= reference_time:
+            return 0
+        else:
+            days = self._timedelta_to_days(new_time - reference_time)
+            value = int(round(days*self.time_divider, 0))
+            if value > OpenPAYGOTokenShared.MAX_ACTIVATION_VALUE:
+                if not force_maximum:
+                    raise Exception('TOO_MANY_DAYS_TO_ACTIVATE')
+                else:
+                    # Will need to be activated again after those days
+                    return OpenPAYGOTokenShared.MAX_ACTIVATION_VALUE
+            return value
+
+    @staticmethod
+    def _timedelta_to_days(this_timedelta):
+        return this_timedelta.days + (this_timedelta.seconds / 3600 / 24)
+
+    @staticmethod
+    def _format_token(token):
+        token = str(token)
+        if len(token) < 9:
+            token = '0' * (9 - len(token)) + token
+        return token

openpaygo/token_decode.py

@@ -134,6 +134,8 @@ def get_activation_value_count_and_type_from_token(
 
     @classmethod
     def _count_is_valid(cls, count, last_count, value, type, used_counts):
+        if used_counts is None:
+            used_counts = []
         if value == OpenPAYGOTokenShared.COUNTER_SYNC_VALUE:
             if count > (last_count - cls.MAX_TOKEN_JUMP):
                 return True
@@ -147,7 +149,7 @@ def _count_is_valid(cls, count, last_count, value, type, used_counts):
 
     @classmethod
     def update_used_counts(cls, past_used_counts, value, new_count, type):
-        if not past_used_counts:
+        if past_used_counts is None:
             return None
         highest_count = max(past_used_counts) if past_used_counts else 0
         if new_count > highest_count:

openpaygo/token_shared.py

@@ -61,14 +61,31 @@ def generate_starting_code(cls, key):
 
     @classmethod
     def load_secret_key_from_hex(cls, secret_key):
+        if isinstance(secret_key, (bytes, bytearray)):
+            secret_key_bytes = bytes(secret_key)
+            if len(secret_key_bytes) != 16:
+                raise ValueError(
+                    "The secret key provided is not correctly formatted, it should be 16 "
+                    "bytes. "
+                )
+            return secret_key_bytes
+
         try:
-            return codecs.decode(secret_key, "hex")
+            decoded = codecs.decode(secret_key, "hex")
         except Exception:
             raise ValueError(
                 "The secret key provided is not correctly formatted, it should be 32 "
                 "hexadecimal characters. "
             )
 
+        if len(decoded) != 16:
+            raise ValueError(
+                "The secret key provided is not correctly formatted, it should be 32 "
+                "hexadecimal characters. "
+            )
+
+        return decoded
+
     @classmethod
     def _convert_to_29_5_bits(cls, source):
         mask = ((1 << (32 - 2 + 1)) - 1) << 2

test/README.md

@@ -0,0 +1,17 @@
+# OpenPAYGO-Token Test
+
+To run the test, first, create and activate a virtual environment. Then install the module and run the test:
+
+```
+$ cd tests
+$ python -m venv env
+$ source env/bin/activate
+$ python -m pip install -e ..
+$ python simple_scenario_test.py
+```
+
+You can do it for full test procedure as well:
+
+```
+$ python full_test_procedure.py
+```