-
-
Notifications
You must be signed in to change notification settings - Fork 12
Porting tests from archived repository of OpenPAYGO-Token #27
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
8cb6abe
1890400
df6e66d
5969ca4
b982f47
90c9b2a
1400dd4
8605868
c47bad5
4cb3518
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,6 @@ | ||
| *.pyc | ||
| dist | ||
| openpaygo.egg-info | ||
| .DS_store | ||
| .DS_store | ||
| __pycache__ | ||
| venv |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -146,10 +146,28 @@ 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 | ||
|
|
||
|
Comment on lines
+149
to
+158
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why do you need to change code in the |
||
| 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 | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| from .device_simulator import DeviceSimulator as DeviceSimulator | ||
| from .server_simulator import SingleDeviceServerSimulator as SingleDeviceServerSimulator | ||
|
|
||
| __all__ = ["DeviceSimulator", "SingleDeviceServerSimulator"] | ||
|
Comment on lines
+1
to
+4
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I don't think these "Simulators" make sense as part of the core module. I know the legacy repository had it that way, but it wasn't ported to Could you move the |
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,162 @@ | ||
| 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) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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: | ||
|
Comment on lines
-150
to
+152
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why is this change required? It seem less "pythonic". Looks like you want to continue processing, when |
||
| return None | ||
| highest_count = max(past_used_counts) if past_used_counts else 0 | ||
| if new_count > highest_count: | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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: | ||
|
|
||
| ```bash | ||
| 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: | ||
|
|
||
| ```bash | ||
| python full_test_procedure.py | ||
| ``` |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please don't change this file. It's centrally managed for all our repositories 😉
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ahh I was getting linting error here for md I believe or smthg that's why had to change that as tests were failing but will see how I can fix it