tests/1.8/stress/stream/pass/string-large/test.py

   1 #!/usr/bin/env python3
   2
   3 # Expects either "prepare" or "clean" argument
   4
   5 import sys
   6 import os
   7
   8 def arg_err():
   9         print('Please specify "prepare <SIZE>" or "clean <SIZE>" as first argument.')
  10         sys.exit(-1)
  11
  12 if (len(sys.argv) <= 1):
  13         arg_err()
  14 mode = sys.argv[1]
  15 if (len(sys.argv) <= 2):
  16         arg_err()
  17 array_len = int(sys.argv[2])
  18
  19 tracedir_name = './trace-' + str(array_len) + '/'
  20 output_metadata = tracedir_name + 'metadata'
  21 output_stream = tracedir_name + 'stream'
  22
  23
  24 def write_binary(f, arr):
  25         f.write(bytes(arr))
  26
  27 def generate_metadata(array_len):
  28         # Generate metadata
  29         metadata_str = \
  30         ('/* CTF 1.8 */\n'
  31         '\n'
  32         'typealias integer {{ size = 8; align = 8; signed = false; base = 10; }} := uint8_t;\n'
  33         'typealias integer {{ size = 32; align = 32; signed = false; base = hex; }} := uint32_t;\n'
  34         '\n'
  35         'trace {{\n'
  36         '       major = 0;\n'
  37         '       minor = 0;\n'
  38         '       uuid = "2a6422d0-6cee-11e0-8c08-cb07d7b3a564";\n'
  39         '       byte_order = le;\n'
  40         '}};\n'
  41         '\n'
  42         'event {{\n'
  43         '       name = myevent;\n'
  44         '       fields := struct {{\n'
  45         '               string mystring;\n'
  46         '       }};\n'
  47         '}};\n').format(array_len)
  48         metadata_f = open(output_metadata, 'w')
  49         metadata_f.write(metadata_str)
  50         metadata_f.close()
  51
  52 def generate_stream(array_len):
  53         # Generate stream
  54         string_content = [ 0x42, ]
  55         string_end = [ 0x0, ]
  56
  57         stream_f = open(output_stream, 'wb')
  58         # generate trace content filled with 1-byte packets
  59         for i in range(array_len - 1):
  60                 write_binary(stream_f, string_content)
  61         write_binary(stream_f, string_end)
  62         stream_f.close()
  63
  64 def test_prepare():
  65         print('Preparing test for large string length ' + str(array_len) + ' bytes')
  66         os.mkdir(tracedir_name)
  67         generate_metadata(array_len)
  68         generate_stream(array_len)
  69
  70 def test_clean():
  71         print('Cleaning up test for large string length ' + str(array_len) + ' bytes')
  72         try:
  73                 os.remove(output_metadata)
  74         except:
  75                 pass
  76         try:
  77                 os.remove(output_stream)
  78         except:
  79                 pass
  80         try:
  81                 os.rmdir(tracedir_name)
  82         except:
  83                 pass
  84
  85 if (mode == "prepare"):
  86         test_clean()
  87         test_prepare()
  88 elif (mode == "clean"):
  89         test_clean()
  90 else:
  91         arg_err()