tests/1.8/stress/stream/pass/array-2GB/test.py

   1 #!/usr/bin/env python3
   2
   3 # Expects either "prepare" or "clean" argument
   4
   5 import sys
   6 import os
   7
   8 output_metadata = './metadata'
   9 output_stream = './stream'
  10
  11 def arg_err():
  12         print('Please specify "prepare" or "clean" as first argument.')
  13         sys.exit(-1)
  14
  15 if (len(sys.argv) <= 1):
  16         arg_err()
  17 mode = sys.argv[1]
  18
  19 metadata_str = \
  20 '/* CTF 1.8 */\n' \
  21 '\n' \
  22 'typealias integer { size = 8; align = 8; signed = false; base = 10; } := uint8_t;\n' \
  23 'typealias integer { size = 32; align = 32; signed = false; base = hex; } := uint32_t;\n' \
  24 '\n' \
  25 'trace {\n' \
  26 '       major = 0;\n' \
  27 '       minor = 0;\n' \
  28 '       uuid = "2a6422d0-6cee-11e0-8c08-cb07d7b3a564";\n' \
  29 '       byte_order = le;\n' \
  30 '       packet.header := struct {\n' \
  31 '               uint32_t magic;\n' \
  32 '               uint8_t uuid[16];\n' \
  33 '       };\n' \
  34 '};\n' \
  35 '\n' \
  36 'event {\n' \
  37 '       name = seq;\n' \
  38 '       fields := struct {\n' \
  39 '               uint8_t field[2147483648]; /* 2GB array */\n' \
  40 '       };\n' \
  41 '};\n'
  42
  43 stream_packet_header = [
  44   0xC1, 0x1F, 0xFC, 0xC1,       # magic
  45   0x2A, 0x64, 0x22, 0xD0, 0x6C, 0xEE, 0x11, 0xE0,
  46   0x8C, 0x08, 0xCB, 0x07, 0xD7, 0xB3, 0xA5, 0x64, # uuid
  47 ]
  48
  49 array_byte = [ 0x00, ]
  50
  51 def write_binary(f, arr):
  52         f.write(bytes(arr))
  53
  54 def generate_metadata():
  55         # Generate metadata
  56         metadata_f = open(output_metadata, 'w')
  57         metadata_f.write(metadata_str)
  58         metadata_f.close()
  59
  60 def generate_stream():
  61         # Generate stream
  62         stream_f = open(output_stream, 'wb')
  63         write_binary(stream_f, stream_packet_header)
  64         # generate 2GB array as event content
  65         os.ftruncate(stream_f.fileno(), 2147483648)
  66         stream_f.close()
  67
  68 def test_prepare():
  69         print('Preparing test')
  70         generate_metadata()
  71         generate_stream()
  72
  73 def test_clean():
  74         print('Cleaning up test')
  75         try:
  76                 os.remove(output_metadata)
  77         except:
  78                 pass
  79         try:
  80                 os.remove(output_stream)
  81         except:
  82                 pass
  83
  84 if (mode == "prepare"):
  85         test_prepare()
  86 elif (mode == "clean"):
  87         test_clean()
  88 else:
  89         arg_err()