src/ceph/src/test/admin_socket/osd_requests

   1 #!/usr/bin/env python
   2
   3 import json
   4 import sys
   5
   6 def main():
   7     """
   8     Read json output of admin socket command 'dump_ops_in_flight' from
   9     stdin, and check that it is consistent.
  10     """
  11     read = sys.stdin.read()
  12     records = json.loads(read)
  13
  14     info_types = ['num_ops', 'ops']
  15     assert sorted(records.keys()) == sorted(info_types)
  16     assert(records['num_ops'] == len(records['ops']))
  17
  18     for op in records['ops']:
  19         assert op['description'] is not None
  20         assert op['received_at'] is not None
  21         assert op['age'] is not None
  22         assert op['flag_point'] is not None
  23         if op['client_info']:
  24             assert op['client_info']['client'] is not None
  25             assert op['client_info']['tid'] is not None
  26
  27 if __name__ == '__main__':
  28     main()