gcpdiag.queries.networkmanagement

31@caching.cached_api_call(in_memory=False)
32def run_connectivity_test(
33  project_id: str, src_ip: str, dest_ip: str, dest_port: int, protocol: str
34):
35  """Method to create/run an idempotent connectivity test"""
36  test_id = f'gcpdiag-connectivity-test-{uuid.uuid4()}'
37  # initialize the networkmanagement api
38  networkmanagement = apis.get_api('networkmanagement', 'v1', project_id)
39
40  # test input
41  test_input = {
42    'source': {'ipAddress': src_ip, 'networkType': 'GCP_NETWORK'},
43    'destination': {'ipAddress': dest_ip, 'port': dest_port},
44    'protocol': protocol,
45  }
46
47  create_request = (
48    networkmanagement.projects()
49    .locations()
50    .global_()
51    .connectivityTests()
52    .create(parent=f'projects/{project_id}/locations/global', testId=test_id, body=test_input)
53  ).execute()
54  logging.info('Running a new connectivity test..')
55
56  # Wait a max of 60 seconds to fetch the request_status.
57  count = 0
58  create_status = (
59    networkmanagement.projects()
60    .locations()
61    .global_()
62    .operations()
63    .get(name=create_request['name'])
64    .execute()
65  )
66  while not create_status['done'] and count <= 15:
67    time.sleep(4)
68    create_status = (
69      networkmanagement.projects()
70      .locations()
71      .global_()
72      .operations()
73      .get(name=create_request['name'])
74      .execute()
75    )
76    count += 1
77
78  if create_status['done']:
79    # get the result of the connectivity test
80    res = (
81      networkmanagement.projects()
82      .locations()
83      .global_()
84      .connectivityTests()
85      .get(name=f'projects/{project_id}/locations/global/connectivityTests/{test_id}')
86    )
87    result = res.execute()
88
89    return result
90  else:
91    logging.warning('Timeout running the connectivity test...')
92  return None