[IMP] base_wamas_ubl: simulate partial reception

Author: jbaudoux

base_wamas_ubl/lib/wamas/tests/samples/wamas2wamas_input_wea_partial.wamas

@@ -0,0 +1 @@
+wamas2wamas_input_wea.wamas

base_wamas_ubl/lib/wamas/tests/samples/wamas2wamas_output_wea_partial.wamas

@@ -0,0 +1,6 @@
+WAMAS     ODOO      00000120231220091116WEAKQ0051000130377              HOST 19700101010000                                                  19700101010000              20231220091116                                                                                           0001040         
+WAMAS     ODOO      00000220231220091116WEAPQ0050000130377              000130377              HOST                     000020                    0001151                00000DISPONIBLE          000000001000000000000000     BOUT                                                                     N
+WAMAS     ODOO      00000320231220091116WEAPQ0050000130377              000130377              HOST                     000030                    0001156                00000DISPONIBLE          000000001000000000000000     PET                                                                      N
+WAMAS     ODOO      00000420231220091116WEAPQ0050000130377              000130377              HOST                     000040                    0001160                00000DISPONIBLE          000000001000000000000000     BOUT                                                                     N
+WAMAS     ODOO      00000520231220091116WEAPQ0050000130377              000130377              HOST                     000050                    0001162                00000DISPONIBLE          000000001000000000000000     PET                                                                      N
+WAMAS     ODOO      00000620231220091116WEAPQ0050000130377              000130377              HOST                     000060                    0001176                00000DISPONIBLE          000000001000000000000000     PET                                                                      N

base_wamas_ubl/lib/wamas/tests/test_wamas2wamas.py

@@ -10,21 +10,25 @@ class TestWamas2wamas(unittest.TestCase):
 
     maxDiff = None
 
-    def _test(self, filename):
+    def _test(self, filename, partial_qty=False):
         with file_open(
             file_path("tests/samples/wamas2wamas_input_%s.wamas" % filename)
         ) as infile, file_open(
             file_path("tests/samples/wamas2wamas_output_%s.wamas" % filename)
         ) as outfile:
             str_input = infile.read()
-            output = wamas2wamas(str_input)
+            output = wamas2wamas(str_input, partial_qty=partial_qty)
             expected_output = outfile.read()
             self.assertEqual(output, expected_output)
 
     @freeze_time("2023-12-20 09:11:16")
-    def testWamas2wamas(self):
+    def testWamas2wamas_full(self):
         self._test("wea")
 
+    @freeze_time("2023-12-20 09:11:16")
+    def testWamas2wamas_partial(self):
+        self._test("wea_partial", partial_qty=True)
+
 
 if __name__ == "__main__":
     unittest.main()

base_wamas_ubl/lib/wamas/utils.py

@@ -79,6 +79,10 @@ def get_current_datetime(val=0):
     return datetime.utcnow()
 
 
+def get_quantity_done(quantity, partial=False):
+    return quantity if not partial else 1
+
+
 def _set_string(val, length, dp, **kwargs):
     return str(val or "").ljust(length)[:length]
 
@@ -327,6 +331,9 @@ def generate_wamas_dict(dict_item, grammar, **kwargs):  # noqa: C901
                 args = (kwargs.get("idx_loop", 0),)
             elif df_func == "get_random_str_num":
                 args = (length,)
+            elif df_func == "get_quantity_done":
+                quantity = dict_item.get("BestMng", 0)
+                args = (quantity, kwargs.get("partial_qty"))
             elif "get_date_from_field" in df_func:
                 args = (dict_wamas_out,)
                 args += ast.literal_eval(re.search(r"\((.*?)\)", df_func).group(0))

base_wamas_ubl/lib/wamas/wamas2wamas.py

@@ -5,13 +5,15 @@
 import logging
 from pprint import pformat
 
+from freezegun import freeze_time
+
 from . import const, utils
 from .wamas2ubl import wamas2dict
 
 _logger = logging.getLogger("wamas2wamas")
 
 
-def simulate_response(dict_wamas_in):
+def simulate_response(dict_wamas_in, partial_qty=False):
     res = []
     line_idx = 0
     dict_parent_id = {}
@@ -27,24 +29,29 @@ def simulate_response(dict_wamas_in):
                     dict_parent_id=dict_parent_id,
                     telegram_type_out=telegram_type_out,
                     do_wamas2wamas=True,
+                    partial_qty=partial_qty,
                 )
                 if line:
                     res.append(line)
     return res
 
 
-def wamas2wamas(infile):
+def wamas2wamas(infile, partial_qty=False):
     data = wamas2dict(infile)
     _logger.debug(pformat(data))
-    wamas_lines = simulate_response(data)
+    wamas_lines = simulate_response(data, partial_qty=partial_qty)
     return "\n".join(wamas_lines)
 
 
+@freeze_time("2023-12-20 09:11:16")
 def main():
     parser = argparse.ArgumentParser(
         description="Converts a wamas message into wamas response.",
     )
     parser.add_argument("-v", "--verbose", action="store_true", help="enable debug log")
+    parser.add_argument(
+        "-p", "--partial", action="store_true", help="simulate partial quantity"
+    )
     parser.add_argument(
         "-o", "--output", dest="outputfile", help="write result in this file"
     )
@@ -53,7 +60,7 @@ def main():
     if args.verbose:
         logging.basicConfig(level=logging.DEBUG)
     infile = utils.file_open(args.inputfile).read()
-    res = wamas2wamas(infile)
+    res = wamas2wamas(infile, args.partial)
     if args.outputfile:
         fd = utils.file_open(args.outputfile, "w")
         fd.write(res)

base_wamas_ubl/lib/wamas/wamas_grammar/weapq.py

@@ -142,9 +142,9 @@
             "type": "float",
             "length": 12,
             "dp": 3,
-            "dict_key": "BestMng",
+            "dict_key": False,
             "df_val": False,
-            "df_func": False,
+            "df_func": "get_quantity_done",
         },
         "IvWevp_LiefMngs_Gew": {
             "type": "float",