VED-470: Add tests for content type parsing.

Author: mfjarvis

mesh_processor/tests/test_converter.py

@@ -59,7 +59,7 @@ def test_non_multipart_content_type(self):
         head_archive_response = s3.head_object(Bucket="source-bucket", Key="archive/test-csv-file.csv")
         assert head_archive_response["ResponseMetadata"]["HTTPStatusCode"] == 200
 
-    def test_non_multipart_content_type_no_mesh_metadata(self):
+    def test_non_multipart_content_type_without_mesh_metadata(self):
         s3 = boto3.client("s3", region_name="eu-west-2")
         s3.put_object(
             Bucket="source-bucket",
@@ -149,8 +149,10 @@ def test_multipart_content_type(self):
                 response = s3.get_object(Bucket="destination-bucket", Key="test-csv-file.csv")
                 body = response["Body"].read().decode("utf-8")
                 assert body == "some CSV content"
+                content_type = response["ContentType"]
+                assert content_type == "text/csv"
 
-    def test_multipart_content_type_without_filename_from_headers(self):
+    def test_multipart_content_type_without_filename_in_headers(self):
         cases = [
             (
                 "no filename in header",
@@ -194,6 +196,33 @@ def test_multipart_content_type_without_filename_from_headers(self):
                 body = response["Body"].read().decode("utf-8")
                 assert body == "some CSV content"
 
+    def test_multipart_content_type_without_content_type_in_headers(self):
+        body = "\r\n".join([
+            "",
+            "--12345678",
+            'Content-Disposition: form-data; name="File"; filename="test-csv-file.csv"',
+            "",
+            "some CSV content",
+            "--12345678--",
+            ""
+        ])
+        s3 = boto3.client("s3", region_name="eu-west-2")
+        s3.put_object(
+            Bucket="source-bucket",
+            Key="test-dat-file.dat",
+            Body=body.encode("utf-8"),
+            ContentType="multipart/form-data; boundary=12345678",
+        )
+
+        result = invoke_lambda("test-dat-file.dat")
+        self.assertEqual(result["statusCode"], 200)
+
+        response = s3.get_object(Bucket="destination-bucket", Key="test-csv-file.csv")
+        body = response["Body"].read().decode("utf-8")
+        assert body == "some CSV content"
+        content_type = response["ContentType"]
+        assert content_type == "application/octet-stream"
+
     def test_multipart_content_type_with_unix_line_endings(self):
         body = "\r\n".join([
             "",