From 5a71d09da9cbb79a52636ba8af35f140f838875b Mon Sep 17 00:00:00 2001 From: HaoWei Date: Sat, 4 Feb 2023 02:39:05 +0800 Subject: [PATCH] Add unit test for new methods --- dysession/aws/dynamodb.py | 3 +-- dysession/backends/db.py | 6 ++--- tests/test_aws_dynamodb.py | 54 ++++++++++++++++++++++++++++++++----- tests/test_backend_db.py | 31 +++++++++++++++++++++ tests/test_backend_model.py | 18 ++++++++----- 5 files changed, 94 insertions(+), 18 deletions(-) diff --git a/dysession/aws/dynamodb.py b/dysession/aws/dynamodb.py index d5cffed..896a959 100644 --- a/dysession/aws/dynamodb.py +++ b/dysession/aws/dynamodb.py @@ -137,7 +137,6 @@ def insert_session_item( insert_item = {pk: data.session_key} for key in data: insert_item[key] = data[key] - print(key, data[key]) response = table.put_item( TableName=table_name, @@ -238,4 +237,4 @@ def delete(self, data: SessionDataModel, table_name: Optional[str] = None) -> bo try: delete_session_item(data=data, table_name=table_name) except AssertionError: - pass + raise diff --git a/dysession/backends/db.py b/dysession/backends/db.py index 91a2636..8a57f46 100644 --- a/dysession/backends/db.py +++ b/dysession/backends/db.py @@ -124,13 +124,11 @@ def delete(self, session_key=None): if session_key is None: session_key = self._session_key - print(f"On Delete: session_key: {session_key}") - self.db.delete(self._get_session()) try: - pass + self.db.delete(self._get_session()) except DeleteSessionError: - print(f"On Delete Error: session_key: {session_key}") + pass def load(self) -> SessionDataModel: """ diff --git a/tests/test_aws_dynamodb.py b/tests/test_aws_dynamodb.py index ed81d8c..742b91a 100644 --- a/tests/test_aws_dynamodb.py +++ b/tests/test_aws_dynamodb.py @@ -6,6 +6,7 @@ from dysession.aws.dynamodb import ( check_dynamodb_table_exists, create_dynamodb_table, + delete_session_item, destory_dynamodb_table, get_item, insert_session_item, @@ -362,9 +363,7 @@ def test_get_item_using_not_exist_key(self): ) with self.assertRaises(DynamodbItemNotFound): - resp = get_item( - session_key="not_exist_key", table_name=options["table"] - ) + resp = get_item(session_key="not_exist_key", table_name=options["table"]) # Insert Item @parameterized.expand( @@ -407,9 +406,7 @@ def test_insert_item_with_tablename(self, session_key: str): resp = insert_session_item(data=model) self.assertEqual(resp["ResponseMetadata"]["HTTPStatusCode"], 200) - resp = get_item( - session_key=session_key, table_name=options["table"] - ) + resp = get_item(session_key=session_key, table_name=options["table"]) self.assertIsInstance(resp, SessionDataModel) @parameterized.expand( @@ -454,3 +451,48 @@ def test_insert_item_without_tablename(self, session_key: str): resp = get_item(session_key=session_key) self.assertIsInstance(resp, SessionDataModel) + + @parameterized.expand( + [ + ["aaaaaaaaa"], + ["bbbbbbbbb"], + ["ccccccccc"], + ] + ) + @mock_dynamodb + def test_delete_item_without_tablename(self, session_key: str): + + options = { + "pk": get_config()["PARTITION_KEY_NAME"], + "sk": get_config()["SORT_KEY_NAME"], + "table": "sessions", + "region": "ap-northeast-1", + } + + client = boto3.client("dynamodb", region_name=options["region"]) + try: + check_dynamodb_table_exists(table_name=options["table"], client=client) + except DynamodbTableNotFound: + create_dynamodb_table( + options={ + "pk": options["pk"], + "sk": options["sk"], + "table": options["table"], + }, + client=client, + ) + + model = SessionDataModel(session_key) + model["a"] = 1 + model["b"] = 2 + model["c"] = 3 + model["d"] = 4 + model["e"] = "qwerty" + + resp = insert_session_item(data=model) + self.assertEqual(resp["ResponseMetadata"]["HTTPStatusCode"], 200) + + resp = delete_session_item(data=model) + + with self.assertRaises(DynamodbItemNotFound): + resp = get_item(session_key=session_key) diff --git a/tests/test_backend_db.py b/tests/test_backend_db.py index ac9ba55..8d10d03 100644 --- a/tests/test_backend_db.py +++ b/tests/test_backend_db.py @@ -233,3 +233,34 @@ def test_exist_check_input_type_error_via_dynamodb_controller( with self.assertRaises(TypeError): db.exists(error_input) + + @mock_dynamodb + def test_delete_item_via_dynamodb_controller(self): + + session_key = "test_set_duplicated_datamodel_via_dynamodb_controller" + self.create_dynamodb_table() + + model = SessionDataModel(session_key) + model["a"] = 1 + model[get_config()["TTL_ATTRIBUTE_NAME"]] = int(datetime.now().timestamp()) + 50 + + db = DynamoDB(self.client) + db.set(model, get_config()["DYNAMODB_TABLENAME"]) + query_model = db.get(session_key=session_key) + self.assertEqual(model.a, query_model.a) + + db.delete(model) + + @mock_dynamodb + def test_delete_item_via_dynamodb_controller_raise_error(self): + + session_key = None + self.create_dynamodb_table() + + model = SessionDataModel(session_key) + model["a"] = 1 + model[get_config()["TTL_ATTRIBUTE_NAME"]] = int(datetime.now().timestamp()) + 50 + + db = DynamoDB(self.client) + with self.assertRaises(AssertionError): + db.delete(model) diff --git a/tests/test_backend_model.py b/tests/test_backend_model.py index 5cd60f6..6c4aead 100644 --- a/tests/test_backend_model.py +++ b/tests/test_backend_model.py @@ -89,13 +89,13 @@ def test_get_function_default_value(self): def test_is_empty(self): model = SessionDataModel() - self.assertTrue(model.is_empty) + self.assertFalse(model.is_empty) model["good_key"] = 0 self.assertFalse(model.is_empty) del model["good_key"] - self.assertTrue(model.is_empty) + self.assertFalse(model.is_empty) def test_iter(self): model = SessionDataModel() @@ -105,7 +105,7 @@ def test_iter(self): model["c"] = 1 model["d"] = 1 - self.assertEqual(set(model), set(["a", "b", "c", "d", "session_key"])) + self.assertEqual(set(model), set(["a", "b", "c", "d", "PK"])) def test_items(self): model = SessionDataModel("session_key") @@ -123,7 +123,7 @@ def test_items(self): values.append(v) - self.assertEqual(set(keys), set(["a", "b", "c", "d", "session_key"])) + self.assertEqual(set(keys), set(["a", "b", "c", "d", "PK"])) self.assertEqual(set(values), set([1, 2, 3, 4, "session_key"])) def test_str_magic_method(self): @@ -137,7 +137,7 @@ def test_str_magic_method(self): data = json.loads(str(model)) - for k in ["session_key", "a", "b", "c", "d"]: + for k in ["PK", "a", "b", "c", "d"]: self.assertIn(k, data.keys()) def test_get__session_expiry(self): @@ -152,4 +152,10 @@ def test_not_found_allow_list(self): model[k] with self.assertRaises(KeyError): - self.assertIsNone(model.get(k)) \ No newline at end of file + self.assertIsNone(model.get(k)) + + + def test_get_and_set_session_key(self): + model = SessionDataModel() + model.session_key = "key" + self.assertEqual(model.session_key, "key") \ No newline at end of file