Add test case for a chunk decompression with the cmp_tool

test/cmp_tool/cmp_tool_integration_test.py

@@ -7,6 +7,7 @@ import os
 import math
 import shutil
 from pathlib import Path
+import hashlib
 
 from datetime import datetime
 from datetime import timedelta
@@ -1415,7 +1416,6 @@ def test_rdcu_pkt():
                      assert(f1.read() == f2.read())
 
     finally:
-        pass
         del_directory('TC_FILES')
         del_file(data_file_name)
         del_file(cfg_file_name)
@@ -1427,5 +1427,59 @@ def test_rdcu_pkt():
         del_file(output_prefix2+'_upmodel.dat')
 
 
+def test_chunk_compression():
+    # decompress the test data
+    output_prefix1 = "ref_short_cadence_1"
+    cmp_data_path1 = "test/cmp_tool/ref_short_cadence_1_cmp.cmp"
+    output_prefix2 = "ref_short_cadence_2"
+    cmp_data_path2 = "test/cmp_tool/ref_short_cadence_2_cmp.cmp"
+
+    try:
+        returncode, stdout, stderr = call_cmp_tool(
+            "--binary -d " + cmp_data_path1 + " -o " + output_prefix1)
+
+        assert(stderr == "")
+        assert(stdout == CMP_START_STR_DECMP +
+               "Importing compressed data file %s ... DONE\n" % (cmp_data_path1) +
+               "Decompress data ... DONE\n" +
+               "Write decompressed data to file %s.dat ... DONE\n" % (output_prefix1))
+        assert(returncode == EXIT_SUCCESS)
+
+        with open(output_prefix1 + '.dat', 'rb') as f:
+            sha1 = hashlib.sha1()
+            while True:
+                chunk = f.read(16 * 1024)
+                if not chunk:
+                    break
+                sha1.update(chunk)
+
+        assert(sha1.hexdigest() == "7d8d94d2ac904f9ff4f934bb691b469a7391ce9e")
+
+        returncode, stdout, stderr = call_cmp_tool(
+            "--binary -d " + cmp_data_path2 + " -m " + output_prefix1 + ".dat -o " + output_prefix2)
+
+        assert(stderr == "")
+        assert(stdout == CMP_START_STR_DECMP +
+               "Importing compressed data file %s ... DONE\n" % (cmp_data_path2) +
+               "Importing model file %s.dat ... DONE\n" % (output_prefix1) +
+               "Decompress data ... DONE\n" +
+               "Write decompressed data to file %s.dat ... DONE\n" % (output_prefix2) +
+               "Write updated model to file %s_upmodel.dat ... DONE\n" % (output_prefix2))
+        assert(returncode == EXIT_SUCCESS)
+
+        with open(output_prefix2 + '.dat', 'rb') as f:
+            sha1 = hashlib.sha1()
+            while True:
+                chunk = f.read(16 * 1024)
+                if not chunk:
+                    break
+                sha1.update(chunk)
+
+        assert(sha1.hexdigest() == "28ecc82a0c44ae7a461c26112b00f65b9b54e66a")
+
+    finally:
+        del_file(output_prefix1+'.dat')
+        del_file(output_prefix2+'.dat')
+        del_file(output_prefix2+'_upmodel.dat')
 
 # TODO: random test

test/cmp_tool/gen_test_data.c

+#include <stdio.h>
+#include <string.h>
+
+#include "../bench/ref_short_cadence_1_cmp.h"
+#include "../bench/ref_short_cadence_2_cmp.h"
+
+
+int main(int argc, char *argv[])
+{
+	int i;
+	FILE *fp;
+	size_t s;
+
+	if (argc < 1)
+		return 1;
+
+	for (i = 1; i < argc; i++) {
+		if (strstr(argv[i], "ref_short_cadence_1_cmp")) {
+			fp = fopen(argv[i], "wb");
+			if(!fp)
+				return 1;
+			s = fwrite(ref_short_cadence_1_cmp, 1, ref_short_cadence_1_cmp_len, fp);
+			fclose(fp);
+			if (s!=ref_short_cadence_1_cmp_len)
+				return 1;
+		} else if (strstr(argv[i], "ref_short_cadence_2_cmp")) {
+			fp = fopen(argv[i], "wb");
+			if(!fp)
+				return 1;
+			s = fwrite(ref_short_cadence_2_cmp, 1, ref_short_cadence_2_cmp_len, fp);
+			fclose(fp);
+			if (s!=ref_short_cadence_2_cmp_len)
+				return 1;
+		} else {
+			fprintf(stderr,"Unknown test data\n");
+			return 1;
+		}
+	}
+
+	return 0;
+}

test/cmp_tool/meson.build

 int_test_file = files('cmp_tool_integration_test.py')
 
+gen_test_data = executable('gen_test_data',
+  'gen_test_data.c',
+   c_args : '-Wno-overlength-strings',
+)
+
+test_data = custom_target('gen_test_data',
+  output : ['ref_short_cadence_1_cmp.cmp', 'ref_short_cadence_2_cmp.cmp'],
+  command : [gen_test_data, '@OUTPUT@']
+)
+
 pytest = find_program('pytest', required : false)
 if pytest.found()
   test('cmp_tool Interface Test',
     pytest,
     args : ['--color=yes', '-vvv', int_test_file],
     env: test_env,
-    depends : cmp_tool_exe,
+    depends : [cmp_tool_exe, test_data],
     timeout : 100,
     workdir : meson.project_build_root())
 else