diff --git a/api/openapi_server/__main__.py b/api/openapi_server/__main__.py index 37b24196..de1e1dbf 100644 --- a/api/openapi_server/__main__.py +++ b/api/openapi_server/__main__.py @@ -1,6 +1,8 @@ from openapi_server.app import create_app from openapi_server.configs.mock_aws import AWSMockService from openapi_server.configs.registry import HUUConfigRegistry +from openapi_server.repositories.user_repo import UserRepository +from openapi_server.models.database import DataAccessLayer if __name__ == "__main__": connexion_app = create_app() @@ -15,7 +17,14 @@ match flask_app.environment: case HUUConfigRegistry.DEVELOPMENT: # Use mocked AWS Cognito service, and temporary user pool - with AWSMockService(flask_app): + with AWSMockService(flask_app) as service: + with DataAccessLayer.session() as session: + user_repo = UserRepository(session) + all_emails = [user.email for user in user_repo.get_all_users()] + + for email in all_emails: + service.add_aws_userpool_user(email, "Test!123") + run_app() case HUUConfigRegistry.STAGING: # Use the real AWS Cognito service, and real user pool diff --git a/api/openapi_server/configs/mock_aws.py b/api/openapi_server/configs/mock_aws.py index 387080a3..e1d7d7f9 100644 --- a/api/openapi_server/configs/mock_aws.py +++ b/api/openapi_server/configs/mock_aws.py @@ -70,10 +70,12 @@ def destroy(self): self.app.logger.info("Destroyed fake temporary userpool") def __enter__(self): - self.create() + self.create() + return self def __exit__(self, exc_type, exc_value, traceback): self.destroy() + return self class AWSMockService(): ''' @@ -138,6 +140,29 @@ def create_test_users(self): self.test_users_created = True + def add_aws_userpool_user(self, email, password, attributes=None): + """ + Adds a new user to the temporary user pool with the given username, password, and attributes. + Attributes should be a list of dictionaries, each containing a 'Name' and 'Value' key. + """ + if attributes is None: + attributes = [] + + try: + response = self.app.boto_client.admin_create_user( + UserPoolId=self.app.config["COGNITO_USER_POOL_ID"], + Username=email, + TemporaryPassword=password, + UserAttributes=attributes, + MessageAction='SUPPRESS' + ) + self._auto_signup_user(email) + self.app.logger.info(f"Added user {email} to the temporary user pool") + return response + except Exception as e: + self.app.logger.error(f"Failed to add user {email}: {str(e)}") + raise + def _auto_signup_user(self, email) -> bool: ''' Auto-confirm a new user. Return True if successful and @@ -193,7 +218,9 @@ def stop(self): self.app.logger.info("Stopped mock AWS Cognito service") def __enter__(self): - self.start() + self.start() + return self def __exit__(self, exc_type, exc_value, traceback): - self.stop() \ No newline at end of file + self.stop() + return self \ No newline at end of file diff --git a/api/openapi_server/repositories/user_repo.py b/api/openapi_server/repositories/user_repo.py index 25eb8466..5904396e 100644 --- a/api/openapi_server/repositories/user_repo.py +++ b/api/openapi_server/repositories/user_repo.py @@ -33,6 +33,9 @@ def delete_user(self, user_id: int) -> bool: def get_user(self, email: str) -> User: return self.session.query(User).filter_by(email=email).first() + def get_all_users(self) -> List[User]: + return self.session.query(User).all() + def get_user_id(self, email: str) -> int: return self.session.query(User).filter_by(email=email).first().id