Feat/staking tests #29
Open
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
OjusWiZard
reviewed
Feb 4, 2025
|
|
||
| cls._setup_complete = True | ||
|
|
||
| def assert_service_stopped(self): |
There was a problem hiding this comment.
can we re-use from BaseTestService
tests/test_staking_service.py
Outdated
Comment on lines
287
to
289
| if not staking_program_id or staking_program_id == "no_staking": | ||
| test_instance.logger.info("Service is not using staking") | ||
| return True |
There was a problem hiding this comment.
probably pass the expected staking program in the function parameter, and match that here
tests/test_staking_service.py
Outdated
Comment on lines
192
to
220
| def get_staking_state(self, service_id: int, staking_address: str) -> StakingState: | ||
| """Get the current staking state for a service.""" | ||
| try: | ||
| # Get staking state using ledger API | ||
| state = StakingState( | ||
| self.staking_contract.get_instance( | ||
| ledger_api=self.ledger_api, | ||
| contract_address=staking_address, | ||
| ) | ||
| .functions.getStakingState(service_id) | ||
| .call() | ||
| ) | ||
|
|
||
| self.logger.info(f"Got staking state for service {service_id}: {state}") | ||
| return state | ||
|
|
||
| except Exception as e: | ||
| self.logger.error(f"Failed to get staking state: {e}") | ||
| raise | ||
|
|
||
| def check_service_staking(self, service_id: int, staking_address: str) -> bool: | ||
| """Check if service is properly staked.""" | ||
| try: | ||
| state = self.get_staking_state(service_id, staking_address) | ||
| return state == StakingState.STAKED | ||
|
|
||
| except Exception as e: | ||
| self.logger.error(f"Staking check failed: {e}") | ||
| return False |
There was a problem hiding this comment.
please re-use these functions from operate
|
|
||
| return settings | ||
|
|
||
| class StakingStatusChecker: |
There was a problem hiding this comment.
this entire class is re-inventing the functions that are already present in operate, please import and use them
There was a problem hiding this comment.
please check this again, needed keys for this so had to init it
tests/test_staking_service.py
Outdated
Comment on lines
50
to
132
| def parse_staking_options(output: str, logger: logging.Logger) -> List[Dict]: | ||
| """Parse staking options and their available slots from CLI output.""" | ||
| options = [] | ||
| logger.info("Starting to parse staking options...") | ||
|
|
||
| # Pattern to match option lines with slots | ||
| pattern = r'(\d+)\)\s*(.*?)\s*\(available slots\s*:\s*([∞\d]+)\)' | ||
| matches = re.finditer(pattern, output) | ||
|
|
||
| for match in matches: | ||
| number = int(match.group(1)) | ||
| name = match.group(2).strip() | ||
| slots_str = match.group(3) | ||
| slots = float('inf') if slots_str == '∞' else int(slots_str) | ||
|
|
||
| option = { | ||
| 'number': number, | ||
| 'name': name, | ||
| 'slots': slots | ||
| } | ||
| options.append(option) | ||
| logger.debug(f"Found option: {option}") | ||
|
|
||
| logger.info(f"Found {len(options)} staking options") | ||
| return options | ||
|
|
||
| @staticmethod | ||
| def select_staking_option(options: List[Dict], logger: logging.Logger) -> str: | ||
| """Select option with maximum available slots, never selecting option 1.""" | ||
| if not options: | ||
| logger.warning("No options parsed, defaulting to option 2") | ||
| return "2" | ||
|
|
||
| # Filter out option 1 and options with no slots | ||
| valid_options = [ | ||
| opt for opt in options | ||
| if opt['number'] != 1 and opt['slots'] > 0 | ||
| ] | ||
|
|
||
| if not valid_options: | ||
| logger.warning("No valid options with available slots, defaulting to option 2") | ||
| return "2" | ||
|
|
||
| # Select option with maximum available slots | ||
| selected = max(valid_options, key=lambda x: x['slots']) | ||
| logger.info(f"Selected option {selected['number']}: {selected['name']} with {selected['slots']} slots") | ||
|
|
||
| return str(selected['number']) | ||
|
|
||
| def handle_staking_choice(output: str, logger: logging.Logger) -> str: | ||
| """Handle staking choice based on available slots.""" | ||
| try: | ||
| parser = StakingOptionParser() | ||
| options = parser.parse_staking_options(output, logger) | ||
| selected = parser.select_staking_option(options, logger) | ||
|
|
||
| # Final safety check | ||
| if selected == "1": | ||
| logger.warning("Safety check caught attempt to select option 1, forcing option 2") | ||
| return "2" | ||
|
|
||
| logger.info(f"Final choice: option {selected}") | ||
| return selected | ||
|
|
||
| except Exception as e: | ||
| logger.error(f"Error in staking choice handler: {str(e)}") | ||
| logger.warning("Falling back to option 2") | ||
| return "2" | ||
|
|
||
| def get_staking_config_settings(config_path: str) -> dict: | ||
| """Get config specific settings with updated staking handler.""" | ||
|
|
||
| # Get original settings from main test script | ||
| settings = get_config_specific_settings(config_path) | ||
|
|
||
| # Remove any default staking choice patterns if they exist | ||
| if r"Enter your choice" in settings["prompts"]: | ||
| settings["prompts"].pop(r"Enter your choice", None) | ||
|
|
||
| # Add our custom staking handler with highest priority | ||
| settings["prompts"][r"Enter your choice \(1 - \d+\):"] = handle_staking_choice | ||
|
|
||
| return settings |
There was a problem hiding this comment.
please make these functions the members of StakingBaseTestService
Signed-off-by: OjusWiZard <[email protected]>
Signed-off-by: OjusWiZard <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description
Adds comprehensive testing for service staking functionality, including rewards claiming, termination, and staking state verification. The changes add a new test file
test_staking_service.pyand updates existing test infrastructure to support staking-specific test cases.Key Changes
Added new
test_staking_service.pywith staking-specific test classes and utilities:StakingOptionParserfor handling staking choice promptsStakingStatusCheckerfor verifying staking statesStakingBaseTestServiceextending base test service with staking capabilitiesUpdated
test_run_service.py:Testing Flow
The new testing framework verifies:
Dependencies
>=3.9,<3.12How to Test
poetry run pytest -v tests/test_staking_service.py -s --log-cli-level=INFONotes
TODO Before Merge