Discover services using WS-Discovery
(scope, ptype, address, port, loglevel, capture, timeout,
unicast_num, multicast_num, relates_to)
| 64 | @click.option('--relates-to', '-rt', is_flag=True, |
| 65 | help='Also use RelatesTo tag to recognize incoming messages.') |
| 66 | def discover(scope, ptype, address, port, loglevel, capture, timeout, |
| 67 | unicast_num, multicast_num, relates_to): |
| 68 | "Discover services using WS-Discovery" |
| 69 | |
| 70 | logger = setup_logger("ws-discovery", loglevel) |
| 71 | |
| 72 | probe_types = [] |
| 73 | for type_tuple in ptype: |
| 74 | probe_types.append(QName(type_tuple[0], type_tuple[2], type_tuple[1])) |
| 75 | if len(probe_types) == 0: |
| 76 | probe_types = None |
| 77 | |
| 78 | with discovery(capture, unicast_num, multicast_num, relates_to) as wsd: |
| 79 | scopes = [Scope(scope)] if scope else [] |
| 80 | svcs = wsd.searchServices(types=probe_types, scopes=scopes, |
| 81 | address=address, port=port, timeout=timeout) |
| 82 | print("\nDiscovered:\n") |
| 83 | for service in svcs: |
| 84 | url = urlparse(service.getXAddrs()[0]) |
| 85 | print(" address: %s" % url.netloc) |
| 86 | print(" - %s\n" % "\n - ".join([str(s) for s in service.getScopes()])) |
| 87 | |
| 88 | |
| 89 | @click.command(context_settings=CONTEXT_SETTINGS) |
nothing calls this directly
no test coverage detected