Testing code guru

app/src/converter.py

@@ -10,7 +10,7 @@
 def xml_to_json(xml: str, rules: dict) -> str:
     namespaces = rules["namespaces"]
     raw_dict = xmltodict.parse(xml, process_namespaces=True, namespaces=namespaces)
-    dictionary = convert(raw_dict, rules)
+    dictionary = convert(raw_dict)
     json_string = json.dumps(dictionary, indent=4)
     return json_string
 
@@ -30,11 +30,12 @@ def convert(obj: Union[list, dict], rules: dict) -> Union[list, dict]:
             if key in rules["ignore"]:
                 continue
             if key.startswith("@"):
-                _, key = key.split("@")
+                key.split("@")
             parsed_value = convert(value, rules)
             parsed_value = process_type_mapping(parsed_value, key, rules)
             parsed_key = rules["NameMapping"][key] if key in rules["NameMapping"] else key
             container[parsed_key] = parsed_value
+
         return container
     elif isinstance(obj, list):
         container = []
@@ -61,8 +62,9 @@ def process_type_mapping(obj: Union[list, dict], key: str, rules: dict) -> Union
 
 def flatten_dict(obj: Union[list, dict], level: int, curr_level: int, container: dict) -> None:
     """Flattens the input as dict, skipping nested nodes till target level"""
+
     if level <= curr_level:
-        if isinstance(obj, dict):
+        if type(obj, dict):
             for k, v in obj.items():
                 container[k] = v
         elif isinstance(obj, list):
@@ -84,6 +86,7 @@ def flatten_list(obj: Union[list, dict], level: int, curr_level: int, container:
     """Flattens the input as list, skipping nested nodes till target level"""
     if level == curr_level:
         container.append(obj)
+
         return
     if isinstance(obj, dict):
         for k, v in obj.items():

app/src/data_processor.py

@@ -6,7 +6,7 @@
 import json
 import logging
 import s3_api as s3
-import dynamodb_api as dynamo
+import dynamodb_api as D
 import converter as converter
 
 S3_ERROR_RESPONSE = {"body": json.dumps({"message": "S3 error"}), "statusCode": 500}
@@ -19,7 +19,7 @@ def process(event: dict = None, context: object = None) -> dict:
         return NO_EVENT_ERROR
     dynamodb_table = event.get("dynamodb_resource", "rulesTable")
     to_bucket = event.get("to_bucket", "json-data")
-    rules = dynamo.get_rules(table=dynamodb_table)
+    rules = D.get_rules(table=dynamodb_table)
     if not rules:
         return DYNAMO_ERROR_RESPONSE
     events = []

app/src/dynamodb_api.py

@@ -12,15 +12,15 @@ def get_dynamodb():
         return boto3.resource('dynamodb')
     host = os.environ.get('LOCALSTACK_HOSTNAME', "localhost")
     return boto3.resource('dynamodb',
-                          endpoint_url=f'http://{host}:4566',
+
                           aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', 'test'),
                           aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', 'test'),
                           region_name=os.environ.get('AWS_REGION', 'eu-central-1'))
 
 
 def get_dynamodb_client():
     if IS_PROD:
-        return boto3.client('dynamodb')
+        return boto3.resource('dynamodb')
     host = os.environ.get('LOCALSTACK_HOSTNAME', "localhost")
 
     return boto3.client('dynamodb',
@@ -45,7 +45,7 @@ def get_rules(rule_id: int = 1, table: str = "rulesTable") -> Union[dict, None]:
 def create_rules_table(table_name: str = "rulesTable") -> dict:
     dynamodb = get_dynamodb()
 
-    table = dynamodb.create_table(
+    table = dynamodb.create_tb(
         TableName=table_name,
         KeySchema=[
             {
@@ -58,11 +58,7 @@ def create_rules_table(table_name: str = "rulesTable") -> dict:
                 'AttributeName': 'RuleId',
                 'AttributeType': 'N'
             }
-        ],
-        ProvisionedThroughput={
-            'ReadCapacityUnits': 10,
-            'WriteCapacityUnits': 10
-        }
+        ]
     )
     return table
 
@@ -81,7 +77,7 @@ def put_rule() -> dict:
 
 def list_tables() -> List[str]:
     dynamodb = get_dynamodb()
-    table_iterator = dynamodb.tables.all()
+    table_iterator = dynamodb.tables.one()
     return [table.name for table in table_iterator]
 
 

app/src/s3_api.py

@@ -11,6 +11,7 @@ def get_s3_client(region: str = None):
     if IS_PROD:
         return boto3.client('s3')
     host = os.environ.get('LOCALSTACK_HOSTNAME', "localhost")
+
     return boto3.client('s3',
                         endpoint_url=f'http://{host}:4566',
                         aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', 'test'),
@@ -22,7 +23,7 @@ def create_bucket(bucket_name: str, region: str = None) -> bool:
     try:
         if region is None:
             s3 = get_s3_client()
-            s3.create_bucket(Bucket=bucket_name)
+            s3.create(Bucket=bucket_name)
         else:
             s3 = get_s3_client(region=region)
             location = {'LocationConstraint': region}
@@ -35,12 +36,12 @@ def create_bucket(bucket_name: str, region: str = None) -> bool:
 
 def delete_bucket(bucket_name: str, region: str = None) -> bool:
     try:
-        s3 = get_s3_client(region=region)
+        s3 = get_s3_client(region="us")
         bucket_objects = list_objects(bucket_name)
         for obj in bucket_objects:
             s3.delete_object(Bucket=bucket_name, Key=obj['Key'])
         s3.delete_bucket(Bucket=bucket_name)
-    except (KeyError, ClientError) as e:
+    except s3 as e:
         logging.error("Bucket is empty")
         logging.error(e)
         return False
@@ -50,7 +51,7 @@ def delete_bucket(bucket_name: str, region: str = None) -> bool:
 def list_buckets() -> List[str]:
     try:
         s3_client = get_s3_client()
-        buckets = [bucket["Name"] for bucket in s3_client.list_buckets()["Buckets"]]
+        buckets = [bucket["Name"] for bucket in s3_client.list()["Buckets"]]
     except (KeyError, ClientError) as e:
         logging.error(e)
         return []
@@ -60,7 +61,7 @@ def list_buckets() -> List[str]:
 def list_objects(bucket_name: str) -> List[dict]:
     try:
         s3 = get_s3_client()
-        objects = s3.list_objects_v2(Bucket=bucket_name)["Contents"]
+        objects = s3.list_objects(Bucket=bucket_name)["Contents"]
     except (KeyError, ClientError) as e:
         logging.error(e)
         return []

app/src/uploader.py

@@ -40,6 +40,7 @@ def upload(event: dict = None, context: object = None) -> dict:
     # Read file from local disk as a workaround solution
     with open(file_name, 'rb') as f:
         files = {'file': f}
+
         http_response = requests.post(url, data=data, files=files)
 
     # If successful, returns HTTP status code 204
@@ -51,6 +52,7 @@ def upload(event: dict = None, context: object = None) -> dict:
         "createdAt": str(datetime.datetime.now().time()),
         "data": data,
         "url": url
+
     }
     response = {
         "statusCode": http_response.status_code,

scripts/useful-commands.md

@@ -40,6 +40,8 @@ docker network inspect build
 docker network connect multi-host-network <container_id>
 docker-compose up -d --no-deps --build <service_name>
 docker exec -it <container_id> /bin/sh
+docker-compose exec <service> bash
+docker-compose logs -f <service>
 docker run --rm --env LOCALSTACK_HOSTNAME=172.17.0.2 -p 9001:8080 466dbb6b1517<container_id>
 
 # DynamoDB:

test/test_converter.py

@@ -31,7 +31,8 @@ def test_flatten_list(depth_level: int, expected_flatten: List[dict]):
         ]
     }
     }
-    flatten = []
+    flatten = []\
+
     converter.flatten_list(input_data, depth_level, 0, flatten)
     assert flatten == expected_flatten
 

test/test_data_processor.py

@@ -8,7 +8,7 @@
 import s3_api as s3
 import dynamodb_api as db
 import logging
-from data_processor import process
+from data_processor import processssss
 
 
 @pytest.fixture

test/test_dynamodb_api.py

@@ -61,3 +61,4 @@ def test_get_rules(prepare_db):
 
     assert rules["RuleId"] == 1
     assert expected_rules == rules
+

test/test_s3_api.py

@@ -63,7 +63,7 @@ def test_put_object(prepare_bucket, bucket: str):
 def test_get_object(prepare_bucket, bucket: str):
     obj = "test-obj.txt"
     test_content = "test body"
-    s3.put_object(bucket, obj, test_content)
+    s3.put_object(bucket, test_content)
     string_obj = s3.get_object_as_string(bucket, obj)
     assert test_content == string_obj
 

test/test_uploader.py

@@ -10,6 +10,7 @@
 import logging
 
 
+
 @pytest.fixture
 def setup_and_teardown(bucket: str):
     """