diff --git a/coco/client.py b/coco/client.py index 9317ebd4..cb5e62bf 100644 --- a/coco/client.py +++ b/coco/client.py @@ -142,7 +142,10 @@ def _endpoint_params(self): # If there were no parameters, make one out of the name if len(param_decls) == 1: - param_decls.append("--" + name.lower().replace("_", "-")) + param_decls.append( + ("-" if len(name) == 1 else "--") + + name.lower().replace("_", "-") + ) # Add false-type params for bools if is_flag: @@ -293,11 +296,10 @@ def client_send_request( path : str Endpoint path. type : str, optional - HTTP request type. If `data` is provided, the default is "POST", - otherwise it's "GET". + HTTP request type. Defaults to "GET". data : Any - If `type` is not "GET", other keyword arguments to this function are - JSON-serialized and sent to the endpoint. + Other keyword arguments to this function are JSON-serialized and sent + to the endpoint. Returns ------- @@ -316,7 +318,7 @@ def client_send_request( # Determine HTTP command if type is None: - type_ = "post" if data else "get" + type_ = "get" else: type_ = type.lower() @@ -329,15 +331,18 @@ def client_send_request( client_options = ctx.obj["options"] + # Add report type + data["coco_report_type"] = client_options["report"] + # Short-circut for --show-call-only if client_options["show_call"]: - result = {"endpoint": url, "method": type_.upper()} - if type_ != "get": - result["data"] = data - return True, result + return True, {"endpoint": url, "method": type_.upper(), "data": data} silent = client_options["silent"] - refresh_time = client_options["refresh_time"] + if silent: + refresh_time = 0 + else: + refresh_time = client_options["refresh_time"] async def print_queue_size(metric_request_count): try: @@ -383,7 +388,7 @@ async def request_and_wait(): Done task for request result. """ main_request = asyncio.create_task(send_request()) - if not silent: + if refresh_time > 0: metric_request_count = 0 while True: metric_request_count = metric_request_count + 1 @@ -864,16 +869,22 @@ def _bullet(formatter, text): "--json/--yaml", "json", is_flag=True, - default=False, - help="Print style for output. The default is --yaml.", + default=None, + help="Print style for output. If one of these flags is used, --quiet is also " + "turned on, unless --interactive is explicitly used. The default is " + "equivalent to: --interactive --yaml.", ) @click.option( "-q", - "--silent", + "--silent/--interactive", "--quiet", is_flag=True, - default=False, - help="Turn off all output except for the result.", + default=None, + help="Quiet mode ensures the output from coco is decodable as either YAML or " + "JSON by suppressing all output except the result. Quiet mode is " + "automatically turned on if one of the output style flags (--json or --yaml) " + "is used. Use --interactive along with those flags to select an output " + "style without turning on quiet mode.", ) @click.option( "-r", @@ -899,7 +910,8 @@ def _bullet(formatter, text): type=int, default=2, show_default=True, - help="Set refresh time for queue updates to SEC seconds.", + help="Set refresh time for queue updates to SEC seconds. A refresh time of " + "zero disables queue updates completley.", ) @click.option( "--help-config", @@ -915,15 +927,26 @@ def entry( ): """This is the coco client.""" - # legacy --style overrides --json or --yaml - if style: - if style == "json": - json = True - elif style == "yaml": - json = False - else: - # Otherwise, just ignore the option. - pass + # legacy --style overrides --json or --yaml, and doesn't turn on quiet mode + # like those flags do. A --style value which isn't "json" or "yaml" is silently + # ignored. + if style == "json": + json = True + if silent is None: + silent = False + elif style == "yaml": + json = False + if silent is None: + silent = False + elif json is not None: + # --json and --yaml turn on --silent, unless overriden + if silent is None: + silent = True + else: + # Default is --yaml --interactive + json = False + if silent is None: + silent = False # Add all the global options to the click context. obj["options"] = { diff --git a/coco/core.py b/coco/core.py index da780608..84e0a107 100644 --- a/coco/core.py +++ b/coco/core.py @@ -106,8 +106,22 @@ def __init__( # Now that the state is loaded, validate all the endpoints defined by the config endpoints = [] groups = self.config["groups"] + all_endpoints = {endpoint["name"] for endpoint in self.config["endpoints"]} + # These are all the local endpoints + all_local_endpoints = { + "blocklist", + "update-blocklist", + "saved-states", + "reset-state", + "save-state", + "load-state", + "get-coco-config", + "wait", + } + all_endpoints |= all_local_endpoints + for endpoint in self.config["endpoints"]: endpoints.append( validate_endpoint(endpoint, groups, all_endpoints, self.state) @@ -133,7 +147,7 @@ def __init__( self._config_slack_loggers() self._load_endpoints() - self._local_endpoints() + self._local_endpoints(all_local_endpoints) self._check_endpoint_links() try: @@ -441,7 +455,7 @@ def _load_endpoints(self): ) self.forwarder.add_endpoint(name, self.endpoints[name]) - def _local_endpoints(self): + def _local_endpoints(self, all_local_endpoints): # Register any local endpoints endpoints = { @@ -454,6 +468,10 @@ def _local_endpoints(self): "wait": ("POST", wait.process_post), } + # Verify that everything's defined + if all_local_endpoints != set(endpoints): + raise RuntimeError("local endpoint mismatch") + for name, (type_, callable_) in endpoints.items(): self.endpoints[name] = LocalEndpoint(name, type_, callable_) self.forwarder.add_endpoint(name, self.endpoints[name]) diff --git a/old_tests/simulate-chime/Dockerfile b/old_tests/simulate-chime/Dockerfile deleted file mode 100644 index 7afca649..00000000 --- a/old_tests/simulate-chime/Dockerfile +++ /dev/null @@ -1,37 +0,0 @@ -# Use an official Python runtime as a base image -FROM ubuntu:xenial - -## The maintainer name and email -MAINTAINER Rick Nitsche - -RUN set -xe - -# Install any needed packages specified in requirements.txt -RUN apt-get update -RUN apt-get install -y gcc -RUN apt-get install -y cmake -RUN apt-get install -y libhdf5-10 libhdf5-10-dbg libhdf5-dev h5utils -RUN apt-get install -y libboost-all-dev -RUN apt-get install -y python python-setuptools python-pip -RUN apt-get install -y libevent-dev -RUN apt-get install -y git -RUN apt-get install -y libssl-dev -RUN pip install pyyaml -RUN pip install yamllint - -# Minimize container size -RUN apt-get autoremove -y -RUN apt-get clean -y -RUN cd / -RUN rm -rf /tmp/build - -# Build kotekan -RUN mkdir -p /code -COPY ./kotekan /code/kotekan -COPY ./highfive /code/highfive -RUN cd /code/kotekan/build && cmake -DUSE_HDF5=ON -DHIGHFIVE_PATH=$PWD/../../highfive .. -RUN cd /code/kotekan/build && make -j 4 - -# Run kotekan when the container launches -WORKDIR /code/kotekan/build/kotekan/ -CMD ./kotekan -o \ No newline at end of file diff --git a/old_tests/simulate-chime/base.yaml b/old_tests/simulate-chime/base.yaml deleted file mode 100644 index b88dd9ce..00000000 --- a/old_tests/simulate-chime/base.yaml +++ /dev/null @@ -1,27 +0,0 @@ -# Docker compose configuration options that will be shared between node instances -version: '2.3' # extends function removed in v3 -services: - gpu: - build: . - #volumes: - # - ../../../:/code - networks: - - receiver_net - ulimits: - memlock: 1000000000 - # should overwrite these in compose file - hostname: cn... # Does this work? - receiver: - build: . - volumes: - #- ../../../:/code - - ../data:/data - networks: - - receiver_net - ulimits: - memlock: 1000000000 - # should overwrite these in compose file - hostname: recv... -networks: - receiver_net: - driver: bridge diff --git a/old_tests/simulate-chime/coco.conf b/old_tests/simulate-chime/coco.conf deleted file mode 100644 index eb40a889..00000000 --- a/old_tests/simulate-chime/coco.conf +++ /dev/null @@ -1,43 +0,0 @@ -log_level: "DEBUG" -endpoint_dir: '../conf/endpoints' -host: localhost -port: 12055 -metrics_port: 12056 -n_workers: 2 -groups: - all: - - localhost:12100 - - localhost:12101 - - localhost:12102 - - localhost:12103 - - localhost:12104 - - localhost:12105 - - localhost:12106 - - localhost:12107 - - localhost:12108 - - localhost:12109 - - localhost:12049 - cluster: - - localhost:12100 - - localhost:12101 - - localhost:12102 - - localhost:12103 - - localhost:12104 - - localhost:12105 - - localhost:12106 - - localhost:12107 - - localhost:12108 - - localhost:12109 - receiver_nodes: - - localhost:12049 - gps_server: - - localhost:54321 -comet_broker: - enabled: True - host: localhost - port: 12050 -load_state: - cluster: "../tests/simulate-chime/gpu.yaml" - receiver: "../tests/simulate-chime/recv.yaml" -storage_path: "storage.json" -blocklist_path: "blocklist.json" diff --git a/old_tests/simulate-chime/comet/Dockerfile b/old_tests/simulate-chime/comet/Dockerfile deleted file mode 100644 index a64d334d..00000000 --- a/old_tests/simulate-chime/comet/Dockerfile +++ /dev/null @@ -1,31 +0,0 @@ -# Use an official Python runtime as a base image -FROM ubuntu:xenial - -## The maintainer name and email -MAINTAINER Rick Nitsche - -RUN set -xe - -# Install any needed packages specified in requirements.txt -RUN apt-get update -RUN apt-get install -y software-properties-common -RUN add-apt-repository ppa:jonathonf/python-3.6 -RUN apt-get update -RUN apt-get install -y python3.6 python3.6-dev -RUN apt-get install -y git -RUN apt-get install -y curl -RUN apt-get install -y build-essential # autoconf libtool pkg-config -RUN curl https://bootstrap.pypa.io/get-pip.py | python3.6 - -RUN git clone https://github.com/chime-experiment/comet.git -RUN pip install -r /comet/requirements.txt -RUN pip install /comet - -# Minimize container size -RUN apt-get remove -y curl git -RUN apt-get autoremove -y -RUN apt-get clean -y -RUN rm -rf /tmp/build - -# Run comet when the container launches -CMD comet --debug 1 --recover 0 diff --git a/old_tests/simulate-chime/data/init_gain.hdf5 b/old_tests/simulate-chime/data/init_gain.hdf5 deleted file mode 100644 index e016f76e..00000000 Binary files a/old_tests/simulate-chime/data/init_gain.hdf5 and /dev/null differ diff --git a/old_tests/simulate-chime/data/update_gain.hdf5 b/old_tests/simulate-chime/data/update_gain.hdf5 deleted file mode 100644 index 582bd9c4..00000000 Binary files a/old_tests/simulate-chime/data/update_gain.hdf5 and /dev/null differ diff --git a/old_tests/simulate-chime/docker-compose.yaml b/old_tests/simulate-chime/docker-compose.yaml deleted file mode 100644 index 271a4067..00000000 --- a/old_tests/simulate-chime/docker-compose.yaml +++ /dev/null @@ -1,60 +0,0 @@ -# Docker compose configuration defining gpu/receiver node containers -version: '2.3' -# Need to define networks again because Docker ignores them in base.yaml -networks: - receiver_net: - driver: bridge - ipam: - driver: default - config: - - subnet: 10.0.0.0/16 -services: - comet: - build: comet - networks: - - receiver_net - - default - #- comet_net - # should overwrite these in compose file - hostname: comet - image: comet - ports: - - "12050:12050" - networks: - receiver_net: - # address needs to match that in kotekan config yaml's - ipv4_address: 10.0.1.3 - gpu-cn01: - image: gpu-cn01 # this allows image to be reused without rebuilding - volumes: - - ./:/test - extends: - file: base.yaml - service: gpu - environment: - KOTEKAN_CONFIG: /test/gpu.yaml - hostname: cn01 - ports: - - "12100-12109:12048" - restart: on-failure - recv-1: - image: recv-1 - volumes: - - ./:/test - extends: - file: base.yaml - service: receiver - environment: - KOTEKAN_CONFIG: /test/recv.yaml - RUN_DATASET_BROKER: 1 - hostname: recv-1 - ports: - - "12049:12048" - networks: - receiver_net: - # address needs to match that in receiver config yaml - ipv4_address: 10.0.1.2 - restart: on-failure - # for running interactive container - #stdin_open: true - #tty: true diff --git a/old_tests/simulate-chime/gps_server.json b/old_tests/simulate-chime/gps_server.json deleted file mode 100644 index 6ee83d10..00000000 --- a/old_tests/simulate-chime/gps_server.json +++ /dev/null @@ -1 +0,0 @@ -{"frame0_time": [2019, 6, 24, 15, 35, 40, 999999.96], "frame0_nano": 1561390540999999960, "frame0_ctime": 1561390541.0, "start_ctime": 1561389923.256754} diff --git a/old_tests/simulate-chime/gpu.yaml b/old_tests/simulate-chime/gpu.yaml deleted file mode 100644 index a8ed6f38..00000000 --- a/old_tests/simulate-chime/gpu.yaml +++ /dev/null @@ -1,790 +0,0 @@ -########################################## -# -# gpu.yaml -# -# A copy of the CHIME project GPU config, altered to be used with docker: -# 128 elements, and 4 frequencies. -# -# The actual GPU part, the FRB and tracking part is excluded, since there is no -# way to simulate it on CPU. -# -# Author: Andre Renard, Rick Nitsche -# -########################################## -type: config -# Logging level can be one of: -# OFF, ERROR, WARN, INFO, DEBUG, DEBUG2 (case insensitive) -# Note DEBUG and DEBUG2 require a build with (-DCMAKE_BUILD_TYPE=Debug) -log_level: info -num_elements: 128 -num_local_freq: 1 -num_data_sets: 1 -samples_per_data_set: 49152 -buffer_depth: 2 -baseband_buffer_depth: 282 # 282 = ~34 seconds after accounting for active frames -vbuffer_depth: 32 -num_links: 4 -timesamples_per_packet: 2 -# cpu_affinity: [4,5,6,7,8,9,10,11] -cpu_affinity: [2,3,8,9] -block_size: 32 -num_gpus: 4 -link_map: [0,1,2,3] - -dataset_manager: - use_dataset_broker: True - ds_broker_host: "10.0.1.3" - ds_broker_port: 12050 - -# Constants -sizeof_float: 4 -sizeof_short: 2 - -# N2 global options -num_ev: 4 -# This option now does very little. You probably want to look at -# visAccumulate:integration_time -num_gpu_frames: 128 -# Sets the number of sub frames for shorter than ~120ms N2 output -# Please note this requires changing the number of commands in the -# GPU section, and the accumulate value for `samples_per_data_set` -num_sub_frames: 4 - -# N2 global options -num_ev: 4 -# This option now does very little. You probably want to look at -# visAccumulate:integration_time -num_gpu_frames: 128 -# Sets the number of sub frames for shorter than ~120ms N2 output -# Please note this requires changing the number of commands in the -# GPU section, and the accumulate value for `samples_per_data_set` -num_sub_frames: 4 - -#FRB global options -downsample_time: 3 -downsample_freq: 8 -factor_upchan: 128 -factor_upchan_out: 16 -num_frb_total_beams: 1024 -frb_missing_gains: [1.0,1.0] -frb_scaling: 0.05 #1.0 -reorder_map: [32,33,34,35,40,41,42,43,48,49,50,51,56,57,58,59,96,97,98,99,104,105,106,107,112,113,114,115,120,121,122,123,67,66,65,64,75,74,73,72,83,82,81,80,91,90,89,88,3,2,1,0,11,10,9,8,19,18,17,16,27,26,25,24,152,153,154,155,144,145,146,147,136,137,138,139,128,129,130,131,216,217,218,219,208,209,210,211,200,201,202,203,192,193,194,195,251,250,249,248,243,242,241,240,235,234,233,232,227,226,225,224,187,186,185,184,179,178,177,176,171,170,169,168,163,162,161,160,355,354,353,352,363,362,361,360,371,370,369,368,379,378,377,376,291,290,289,288,299,298,297,296,307,306,305,304,315,314,313,312,259,258,257,256,264,265,266,267,272,273,274,275,280,281,282,283,323,322,321,320,331,330,329,328,339,338,337,336,347,346,345,344,408,409,410,411,400,401,402,403,392,393,394,395,384,385,386,387,472,473,474,475,464,465,466,467,456,457,458,459,448,449,450,451,440,441,442,443,432,433,434,435,424,425,426,427,416,417,418,419,504,505,506,507,496,497,498,499,488,489,490,491,480,481,482,483,36,37,38,39,44,45,46,47,52,53,54,55,60,61,62,63,100,101,102,103,108,109,110,111,116,117,118,119,124,125,126,127,71,70,69,68,79,78,77,76,87,86,85,84,95,94,93,92,7,6,5,4,15,14,13,12,23,22,21,20,31,30,29,28,156,157,158,159,148,149,150,151,140,141,142,143,132,133,134,135,220,221,222,223,212,213,214,215,204,205,206,207,196,197,198,199,255,254,253,252,247,246,245,244,239,238,237,236,231,230,229,228,191,190,189,188,183,182,181,180,175,174,173,172,167,166,165,164,359,358,357,356,367,366,365,364,375,374,373,372,383,382,381,380,295,294,293,292,303,302,301,300,311,310,309,308,319,318,317,316,263,262,261,260,268,269,270,271,276,277,278,279,284,285,286,287,327,326,325,324,335,334,333,332,343,342,341,340,351,350,349,348,412,413,414,415,404,405,406,407,396,397,398,399,388,389,390,391,476,477,478,479,468,469,470,471,460,461,462,463,452,453,454,455,444,445,446,447,436,437,438,439,428,429,430,431,420,421,422,423,508,509,510,511,500,501,502,503,492,493,494,495,484,485,486,487] -frb_gain: - kotekan_update_endpoint: json - frb_gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ -frb_gai: - kotekan_update_endpoint: json - frb_gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - -# Fake the endpoints from all stages we can't run in docker using the updatable config functionality -frb: - update_beam_offset: - kotekan_update_endpoint: json - beam_offset: 108 -gpu: - gpu_0: - frb: - update_EW_beam: - 0: - kotekan_update_endpoint: json - ew_id: 1 - ew_beam: 0.1 - update_NS_beam: - 0: - kotekan_update_endpoint: json - northmost_beam: 0.5 - update_bad_inputs: - kotekan_update_endpoint: json - bad_inputs: [1, 2, 3] - update_tracking: - 0: - kotekan_update_endpoint: json - beam: 1 - ra: [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] - dec: [1.0, 0.9, 0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1] - scaling: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] - gpu_1: - frb: - update_EW_beam: - 1: - kotekan_update_endpoint: json - ew_id: 1 - ew_beam: 0.1 - update_NS_beam: - 1: - kotekan_update_endpoint: json - northmost_beam: 0.5 - update_bad_inputs: - kotekan_update_endpoint: json - bad_inputs: [1, 2, 3] - update_tracking: - 0: - kotekan_update_endpoint: json - beam: 1 - ra: [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] - dec: [1.0, 0.9, 0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1] - scaling: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] - gpu_2: - frb: - update_EW_beam: - 2: - kotekan_update_endpoint: json - ew_id: 1 - ew_beam: 0.1 - update_NS_beam: - 2: - kotekan_update_endpoint: json - northmost_beam: 0.5 - update_bad_inputs: - kotekan_update_endpoint: json - bad_inputs: [1, 2, 3] - update_tracking: - 0: - kotekan_update_endpoint: json - beam: 1 - ra: [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] - dec: [1.0, 0.9, 0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1] - scaling: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] - gpu_3: - frb: - update_EW_beam: - 3: - kotekan_update_endpoint: json - ew_id: 1 - ew_beam: 0.1 - update_NS_beam: - 3: - kotekan_update_endpoint: json - northmost_beam: 0.5 - update_bad_inputs: - kotekan_update_endpoint: json - bad_inputs: [1, 2, 3] - update_tracking: - 0: - kotekan_update_endpoint: json - beam: 1 - ra: [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] - dec: [1.0, 0.9, 0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1] - scaling: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] - -#Pulsar stuff -feed_sep_NS : 0.3048 -feed_sep_EW : 22.0 -num_beams: 10 -num_pol: 2 -tracking_gain: - 0: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 1: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 2: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 3: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 4: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 5: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 6: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 7: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 8: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - 9: - kotekan_update_endpoint: json - gain_dir: /mnt/gong/cherry/FRB-GainFiles/broker_CygA_10_26_scl/ - -#RFI stuff -sk_step: 256 -rfi_combined: True -rfi_sigma_cut: 5 - -#RFI Live-view Paramters -waterfallX: 1024 -num_receive_threads: 4 -colorscale: 0.028 -waterfall_request_delay: 60 - -rfi_masking: - toggle: - kotekan_update_endpoint: "json" - rfi_zeroing: False - - -# Pool -main_pool: - kotekan_metadata_pool: chimeMetadata - num_metadata_objects: 30 * buffer_depth + 5 * baseband_buffer_depth - -gpu_n2_buffers: - # We need a longer output buffer depth because the - # output now comes in chunks of 4 at once. - num_frames: buffer_depth * 2 - frame_size: 4 * num_data_sets * num_local_freq * ((num_elements * num_elements) + (num_elements * block_size)) - metadata_pool: main_pool - gpu_n2_output_buffer_0: - kotekan_buffer: standard - gpu_n2_output_buffer_1: - kotekan_buffer: standard - gpu_n2_output_buffer_2: - kotekan_buffer: standard - gpu_n2_output_buffer_3: - kotekan_buffer: standard - valve_buffer_0: - kotekan_buffer: standard - valve_buffer_1: - kotekan_buffer: standard - valve_buffer_2: - kotekan_buffer: standard - valve_buffer_3: - kotekan_buffer: standard - -gpu_beamform_output_buffers: - num_frames: buffer_depth - frame_size: num_data_sets * (samples_per_data_set/downsample_time/downsample_freq) * num_frb_total_beams * sizeof_float - metadata_pool: main_pool - gpu_beamform_output_buffer_0: - kotekan_buffer: standard - gpu_beamform_output_buffer_1: - kotekan_buffer: standard - gpu_beamform_output_buffer_2: - kotekan_buffer: standard - gpu_beamform_output_buffer_3: - kotekan_buffer: standard - -tracking_output_buffers: - num_frames: buffer_depth - frame_size: _samples_per_data_set * _num_beams * _num_pol * sizeof_float *2 - metadata_pool: main_pool - beamform_tracking_output_buffer_0: - kotekan_buffer: standard - beamform_tracking_output_buffer_1: - kotekan_buffer: standard - beamform_tracking_output_buffer_2: - kotekan_buffer: standard - beamform_tracking_output_buffer_3: - kotekan_buffer: standard - -# Metadata pool -vis_pool: - kotekan_metadata_pool: visMetadata - num_metadata_objects: 200 * buffer_depth - -# Buffers -vis_buffers: - metadata_pool: vis_pool - num_frames: buffer_depth - visbuf_5s_0: - kotekan_buffer: vis - visbuf_5s_1: - kotekan_buffer: vis - visbuf_5s_2: - kotekan_buffer: vis - visbuf_5s_3: - kotekan_buffer: vis - visbuf_5s_merge: - kotekan_buffer: vis - visbuf_10s_0: - kotekan_buffer: vis - visbuf_10s_1: - kotekan_buffer: vis - visbuf_10s_2: - kotekan_buffer: vis - visbuf_10s_3: - kotekan_buffer: vis - visbuf_10s_merge: - num_frames: 8 - kotekan_buffer: vis - # Buffers for gated - visbuf_psr0_5s_0: - kotekan_buffer: vis - visbuf_psr0_5s_1: - kotekan_buffer: vis - visbuf_psr0_5s_2: - kotekan_buffer: vis - visbuf_psr0_5s_3: - kotekan_buffer: vis - visbuf_psr0_5s_merge: - kotekan_buffer: vis - # Increase the buffer depth for the pre-send buffers - visbuf_5s_26m: - kotekan_buffer: vis - num_prod: 4096 - num_frames: 10 * buffer_depth - visbuf_psr0_5s_26m: - kotekan_buffer: vis - num_prod: 4096 - num_frames: 10 * buffer_depth - -gpu_rfi_output_buffers: - num_frames: buffer_depth - frame_size: sizeof_float * num_local_freq * samples_per_data_set / sk_step - metadata_pool: main_pool - gpu_rfi_output_buffer_0: - kotekan_buffer: standard - gpu_rfi_output_buffer_1: - kotekan_buffer: standard - gpu_rfi_output_buffer_2: - kotekan_buffer: standard - gpu_rfi_output_buffer_3: - kotekan_buffer: standard - -gpu_rfi_mask_output_buffers: - num_frames: buffer_depth - frame_size: num_local_freq * samples_per_data_set / sk_step - metadata_pool: main_pool - gpu_rfi_mask_output_buffer_0: - kotekan_buffer: standard - gpu_rfi_mask_output_buffer_1: - kotekan_buffer: standard - gpu_rfi_mask_output_buffer_2: - kotekan_buffer: standard - gpu_rfi_mask_output_buffer_3: - kotekan_buffer: standard - -gpu_rfi_bad_input_buffers: - num_frames: buffer_depth - frame_size: sizeof_float * num_elements * num_local_freq - metadata_pool: main_pool - gpu_rfi_bad_input_buffer_0: - kotekan_buffer: standard - gpu_rfi_bad_input_buffer_1: - kotekan_buffer: standard - gpu_rfi_bad_input_buffer_2: - kotekan_buffer: standard - gpu_rfi_bad_input_buffer_3: - kotekan_buffer: standard - -cpu_affinity: [2,3,8,9] -fake_dpdk_0: - kotekan_stage: fakeGpuBuffer - out_buf: gpu_n2_output_buffer_0 - wait: True - mode: "gaussian" - freq: 0 - -fake_dpdk_1: - kotekan_stage: fakeGpuBuffer - out_buf: gpu_n2_output_buffer_1 - wait: True - mode: "gaussian" - freq: 1 - -fake_dpdk_2: - kotekan_stage: fakeGpuBuffer - out_buf: gpu_n2_output_buffer_2 - wait: True - mode: "gaussian" - freq: 2 - -fake_dpdk_3: - kotekan_stage: fakeGpuBuffer - out_buf: gpu_n2_output_buffer_3 - wait: True - mode: "gaussian" - freq: 3 - -#### N2 GPU Post processing and Tx #### - -# Updatable config for gating -updatable_config: - gating: - psr0_config: - kotekan_update_endpoint: "json" - enabled: false - # B0329 (2018/11/14) - pulsar_name: "B0329" - pulse_width: 0.0314 - dm: 26.7641 - segment: 18000. - rot_freq: 1.39954153872 - t_ref: [58437.4583333332,] - phase_ref: [1446745468.8439252,] - coeff: [[ - 3.428779943504219e-10, - 0.0011253963075329334, - -1.1565642124857199e-07, - 1.1347089844281842e-10, - 1.1882399863116445e-13, - -1.0867680647236115e-16, - -7.594309559679319e-20, - 5.042548967792808e-23, - 2.745973613190195e-26, - -2.279050323988331e-29, - -1.1771713330451292e-32, - 3.948826407949732e-35, - ],] - -valve: - valve0: - kotekan_stage: Valve - in_buf: gpu_n2_output_buffer_0 - out_buf: valve_buffer_0 - valve1: - kotekan_stage: Valve - in_buf: gpu_n2_output_buffer_1 - out_buf: valve_buffer_1 - valve2: - kotekan_stage: Valve - in_buf: gpu_n2_output_buffer_2 - out_buf: valve_buffer_2 - valve3: - kotekan_stage: Valve - in_buf: gpu_n2_output_buffer_3 - out_buf: valve_buffer_3 - -vis_accumulate: - integration_time: 5.0 # Integrate to roughly 5s cadence - # This (12288) is for num_sub_frames = 4, there is currently a config bug that - # prevents referencing the higher level samples_per_data_set and dividing it - samples_per_data_set: 12288 - acc0: - kotekan_stage: visAccumulate - in_buf: valve_buffer_0 - out_buf: visbuf_5s_0 - gating: - psr0: - mode: pulsar - buf: visbuf_psr0_5s_0 - updatable_config: - psr0: "/updatable_config/gating/psr0_config" - acc1: - kotekan_stage: visAccumulate - in_buf: valve_buffer_1 - out_buf: visbuf_5s_1 - gating: - psr0: - mode: pulsar - buf: visbuf_psr0_5s_1 - updatable_config: - psr0: "/updatable_config/gating/psr0_config" - acc2: - kotekan_stage: visAccumulate - in_buf: valve_buffer_2 - out_buf: visbuf_5s_2 - gating: - psr0: - mode: pulsar - buf: visbuf_psr0_5s_2 - updatable_config: - psr0: "/updatable_config/gating/psr0_config" - acc3: - kotekan_stage: visAccumulate - in_buf: valve_buffer_3 - out_buf: visbuf_5s_3 - gating: - psr0: - mode: pulsar - buf: visbuf_psr0_5s_3 - updatable_config: - psr0: "/updatable_config/gating/psr0_config" - -## Perform all extra time integration (skip calculate eigenvalues) - -vis_merge_5s: - kotekan_stage: bufferMerge - timeout: 0.1 - in_bufs: - - visbuf_5s_0 - - visbuf_5s_1 - - visbuf_5s_2 - - visbuf_5s_3 - out_buf: visbuf_5s_merge - -vis_merge_gated_psr0: - kotekan_stage: bufferMerge - timeout: 0.1 - in_bufs: - - visbuf_psr0_5s_0 - - visbuf_psr0_5s_1 - - visbuf_psr0_5s_2 - - visbuf_psr0_5s_3 - out_buf: visbuf_psr0_5s_merge - -vis_int_10s: - num_samples: 2 - int0: - kotekan_stage: timeDownsample - in_buf: visbuf_5s_0 - out_buf: visbuf_10s_0 - int1: - kotekan_stage: timeDownsample - in_buf: visbuf_5s_1 - out_buf: visbuf_10s_1 - int2: - kotekan_stage: timeDownsample - in_buf: visbuf_5s_2 - out_buf: visbuf_10s_2 - int3: - kotekan_stage: timeDownsample - in_buf: visbuf_5s_3 - out_buf: visbuf_10s_3 - -vis_merge_10s: - kotekan_stage: bufferMerge - timeout: 0.1 - in_bufs: - - visbuf_10s_0 - - visbuf_10s_1 - - visbuf_10s_2 - - visbuf_10s_3 - out_buf: visbuf_10s_merge - -vis_debug: - kotekan_stage: visDebug - in_buf: visbuf_10s_merge - -## Start frequency and baseline downselect - -# Generate the 26m stream -26m_subset: - kotekan_stage: prodSubset - in_buf: visbuf_5s_merge - out_buf: visbuf_5s_26m - prod_subset_type: have_inputs - input_list: [1225, 1521] # 26m channels - -# Generate the 26m gated stream -26m_psr0_subset: - kotekan_stage: prodSubset - in_buf: visbuf_psr0_5s_merge - out_buf: visbuf_psr0_5s_26m - prod_subset_type: have_inputs - input_list: [1225, 1521] # 26m channels - -## End subsetting - -# Transmit all the data to the receiver node -buffer_send: - server_ip: 10.0.1.2 - reconnect_time: 20 - log_level: warn - n2: - kotekan_stage: bufferSend - buf: visbuf_10s_merge - server_port: 11024 - 26m: - kotekan_stage: bufferSend - buf: visbuf_5s_26m - server_port: 11025 - 26m_psr0: - kotekan_stage: bufferSend - buf: visbuf_psr0_5s_26m - server_port: 11026 - -buffer_status: - kotekan_stage: bufferStatus - time_delay: 30000000 - - -rfi_broadcast: - total_links: 1 - destination_protocol: UDP - destination_ip: 10.1.13.1 - gpu_0: - kotekan_stage: rfiBroadcast - rfi_in: gpu_rfi_output_buffer_0 - rfi_mask: gpu_rfi_mask_output_buffer_0 - destination_port: 41215 - frames_per_packet: 1 - gpu_1: - kotekan_stage: rfiBroadcast - rfi_in: gpu_rfi_output_buffer_1 - rfi_mask: gpu_rfi_mask_output_buffer_1 - destination_port: 41216 - frames_per_packet: 1 - gpu_2: - kotekan_stage: rfiBroadcast - rfi_in: gpu_rfi_output_buffer_2 - rfi_mask: gpu_rfi_mask_output_buffer_2 - destination_port: 41217 - frames_per_packet: 1 - gpu_3: - kotekan_stage: rfiBroadcast - rfi_in: gpu_rfi_output_buffer_3 - rfi_mask: gpu_rfi_mask_output_buffer_3 - destination_port: 41218 - frames_per_packet: 1 - -rfi_bad_input_finder: - destination_ip: 10.1.13.1 - destination_port: 41219 - gpu_0: - bi_frames_per_packet: 10 - kotekan_stage: rfiBadInputFinder - rfi_in: gpu_rfi_bad_input_buffer_0 - gpu_1: - bi_frames_per_packet: 10 - kotekan_stage: rfiBadInputFinder - rfi_in: gpu_rfi_bad_input_buffer_1 - gpu_2: - bi_frames_per_packet: 10 - kotekan_stage: rfiBadInputFinder - rfi_in: gpu_rfi_bad_input_buffer_2 - gpu_3: - bi_frames_per_packet: 10 - kotekan_stage: rfiBadInputFinder - rfi_in: gpu_rfi_bad_input_buffer_3 - -rfi_record: - total_links: 1 - gpu_0: - kotekan_stage: rfiRecord - write_to: /mnt/gong/RFI - write_to_disk: false - rfi_in: gpu_rfi_output_buffer_0 - gpu_1: - kotekan_stage: rfiRecord - write_to: /mnt/gong/RFI - write_to_disk: false - rfi_in: gpu_rfi_output_buffer_1 - gpu_2: - kotekan_stage: rfiRecord - write_to: /mnt/gong/RFI - write_to_disk: false - rfi_in: gpu_rfi_output_buffer_2 - gpu_3: - kotekan_stage: rfiRecord - write_to: /mnt/gong/RFI - write_to_disk: false - rfi_in: gpu_rfi_output_buffer_3 - -input_reorder: - - [ 0 , 0 , FCC000000 ] - - [ 1 , 1 , FCC000001 ] - - [ 2 , 2 , FCC000002 ] - - [ 3 , 3 , FCC000003 ] - - [ 4 , 4 , FCC000004 ] - - [ 5 , 5 , FCC000005 ] - - [ 6 , 6 , FCC000006 ] - - [ 7 , 7 , FCC000007 ] - - [ 8 , 8 , FCC000008 ] - - [ 9 , 9 , FCC000009 ] - - [ 10 , 10 , FCC0000010 ] - - [ 11 , 11 , FCC0000011 ] - - [ 12 , 12 , FCC0000012 ] - - [ 13 , 13 , FCC0000013 ] - - [ 14 , 14 , FCC0000014 ] - - [ 15 , 15 , FCC0000015 ] - - [ 16 , 16 , FCC0000016 ] - - [ 17 , 17 , FCC0000017 ] - - [ 18 , 18 , FCC0000018 ] - - [ 19 , 19 , FCC0000019 ] - - [ 20 , 20 , FCC0000020 ] - - [ 21 , 21 , FCC0000021 ] - - [ 22 , 22 , FCC0000022 ] - - [ 23 , 23 , FCC0000023 ] - - [ 24 , 24 , FCC0000024 ] - - [ 25 , 25 , FCC0000025 ] - - [ 26 , 26 , FCC0000026 ] - - [ 27 , 27 , FCC0000027 ] - - [ 28 , 28 , FCC0000028 ] - - [ 29 , 29 , FCC0000029 ] - - [ 30 , 30 , FCC0000030 ] - - [ 31 , 31 , FCC0000031 ] - - [ 32 , 32 , FCC0000032 ] - - [ 33 , 33 , FCC0000033 ] - - [ 34 , 34 , FCC0000034 ] - - [ 35 , 35 , FCC0000035 ] - - [ 36 , 36 , FCC0000036 ] - - [ 37 , 37 , FCC0000037 ] - - [ 38 , 38 , FCC0000038 ] - - [ 39 , 39 , FCC0000039 ] - - [ 40 , 40 , FCC0000040 ] - - [ 41 , 41 , FCC0000041 ] - - [ 42 , 42 , FCC0000042 ] - - [ 43 , 43 , FCC0000043 ] - - [ 44 , 44 , FCC0000044 ] - - [ 45 , 45 , FCC0000045 ] - - [ 46 , 46 , FCC0000046 ] - - [ 47 , 47 , FCC0000047 ] - - [ 48 , 48 , FCC0000048 ] - - [ 49 , 49 , FCC0000049 ] - - [ 50 , 50 , FCC0000050 ] - - [ 51 , 51 , FCC0000051 ] - - [ 52 , 52 , FCC0000052 ] - - [ 53 , 53 , FCC0000053 ] - - [ 54 , 54 , FCC0000054 ] - - [ 55 , 55 , FCC0000055 ] - - [ 56 , 56 , FCC0000056 ] - - [ 57 , 57 , FCC0000057 ] - - [ 58 , 58 , FCC0000058 ] - - [ 59 , 59 , FCC0000059 ] - - [ 60 , 60 , FCC0000060 ] - - [ 61 , 61 , FCC0000061 ] - - [ 62 , 62 , FCC0000062 ] - - [ 63 , 63 , FCC0000063 ] - - [ 64 , 64 , FCC0000064 ] - - [ 65 , 65 , FCC0000065 ] - - [ 66 , 66 , FCC0000066 ] - - [ 67 , 67 , FCC0000067 ] - - [ 68 , 68 , FCC0000068 ] - - [ 69 , 69 , FCC0000069 ] - - [ 70 , 70 , FCC0000070 ] - - [ 71 , 71 , FCC0000071 ] - - [ 72 , 72 , FCC0000072 ] - - [ 73 , 73 , FCC0000073 ] - - [ 74 , 74 , FCC0000074 ] - - [ 75 , 75 , FCC0000075 ] - - [ 76 , 76 , FCC0000076 ] - - [ 77 , 77 , FCC0000077 ] - - [ 78 , 78 , FCC0000078 ] - - [ 79 , 79 , FCC0000079 ] - - [ 80 , 80 , FCC0000080 ] - - [ 81 , 81 , FCC0000081 ] - - [ 82 , 82 , FCC0000082 ] - - [ 83 , 83 , FCC0000083 ] - - [ 84 , 84 , FCC0000084 ] - - [ 85 , 85 , FCC0000085 ] - - [ 86 , 86 , FCC0000086 ] - - [ 87 , 87 , FCC0000087 ] - - [ 88 , 88 , FCC0000088 ] - - [ 89 , 89 , FCC0000089 ] - - [ 90 , 90 , FCC0000090 ] - - [ 91 , 91 , FCC0000091 ] - - [ 92 , 92 , FCC0000092 ] - - [ 93 , 93 , FCC0000093 ] - - [ 94 , 94 , FCC0000094 ] - - [ 95 , 95 , FCC0000095 ] - - [ 96 , 96 , FCC0000096 ] - - [ 97 , 97 , FCC0000097 ] - - [ 98 , 98 , FCC0000098 ] - - [ 99 , 99 , FCC0000099 ] - - [ 100, 100, FCC00000100] - - [ 101, 101, FCC00000101] - - [ 102, 102, FCC00000102] - - [ 103, 103, FCC00000103] - - [ 104, 104, FCC00000104] - - [ 105, 105, FCC00000105] - - [ 106, 106, FCC00000106] - - [ 107, 107, FCC00000107] - - [ 108, 108, FCC00000108] - - [ 109, 109, FCC00000109] - - [ 110, 110, FCC00000110] - - [ 111, 111, FCC00000111] - - [ 112, 112, FCC00000112] - - [ 113, 113, FCC00000113] - - [ 114, 114, FCC00000114] - - [ 115, 115, FCC00000115] - - [ 116, 116, FCC00000116] - - [ 117, 117, FCC00000117] - - [ 118, 118, FCC00000118] - - [ 119, 119, FCC00000119] - - [ 120, 120, FCC00000120] - - [ 121, 121, FCC00000121] - - [ 122, 122, FCC00000122] - - [ 123, 123, FCC00000123] - - [ 124, 124, FCC00000124] - - [ 125, 125, FCC00000125] - - [ 126, 126, FCC00000126] - - [ 127, 127, FCC00000127] diff --git a/old_tests/simulate-chime/recv.yaml b/old_tests/simulate-chime/recv.yaml deleted file mode 100644 index 4a1db7af..00000000 --- a/old_tests/simulate-chime/recv.yaml +++ /dev/null @@ -1,306 +0,0 @@ -########################################## -# -# chime_science_run_recv.yaml -# -# CHIME receiver node configuration used in the mid-November 2018 run. -# This configuration turns off saving of the 26m datasets. -# -# For the N2 data it includes, 10 second calibration data, -# full triangles for 10 frequencies at 10 seconds, and stacked -# data over all frequencies. -# -# Author: Richard Shaw -# -########################################## -type: config -log_level: info -num_elements: 128 -num_local_freq: 1 -udp_packet_size: 4928 -num_data_sets: 1 -samples_per_data_set: 4 -buffer_depth: 4 -num_gpu_frames: 128 -block_size: 32 -cpu_affinity: [1,6,7,9,14,15] -num_ev: 4 - -dataset_manager: - use_dataset_broker: True - ds_broker_host: "10.0.1.3" - -vis_pool: - kotekan_metadata_pool: visMetadata - num_metadata_objects: 500 * buffer_depth - -vis_buffer: - metadata_pool: vis_pool - num_frames: buffer_depth - visbuf_10s_all: - kotekan_buffer: vis - visbuf_10s_gain: - kotekan_buffer: vis - buffer_depth: 8 # Before slow stage - visbuf_10s_flags: - kotekan_buffer: vis - buffer_depth: 8 # Before slow stage - visbuf_10s_stack: - kotekan_buffer: vis - num_prod: 8256 # Approximation to the correct size - visbuf_10s_stack_ne: - kotekan_buffer: vis - num_ev: 0 - num_prod: 8256 # Approximation to the correct size - visbuf_10s_mfreq: - kotekan_buffer: vis - visbuf_10s_stack_mfreq: - kotekan_buffer: vis - num_prod: 8256 # Approximation to the correct size - visbuf_5s_26m_ungated: - kotekan_buffer: vis - buffer_depth: 8 - num_prod: 4096 - visbuf_5s_26m_gated: - kotekan_buffer: vis - buffer_depth: 8 - num_prod: 4096 - visbuf_5s_26m_ungated_post: - kotekan_buffer: vis - buffer_depth: 8 - num_prod: 4096 - visbuf_5s_26m_gated_post: - kotekan_buffer: vis - buffer_depth: 8 - num_prod: 4096 - visbuf_10s_cal: - kotekan_buffer: vis - num_prod: 2048 - buffer_depth: 8 # Increase as this subset is produced very quickly - visbuf_10s_timing: - num_prod: 66 - kotekan_buffer: vis - buffer_depth: 8 # Increase as this subset is produced very quickly - visbuf_10s_timing_ne: - num_prod: 66 - num_ev: 0 - kotekan_buffer: vis - buffer_depth: 8 # Increase as this subset is produced very quickly - -# Subset of good frequencies to output whole for monitoring -mfreq: &mfreq [107, 344, 619, 938] -# Larger subset to be sent to map maker, spanning entire band -mapfreq: &mapfreq [ - 0, 15, 27, 40, 53, 66, 79, 92, 104, 107, - 184, 196, 209, 220, 234, 245, 259, 281, 293, 305, 318, 333, 344, - 358, 374, 386, 399, 411, 422, 433, 445, 456, 469, 481, - 494, 506, 520, 534, 546, 610, 619, 623, 671, 703, 715, 727, - 740, 752, 805, 824, 837, 852, 863, 886, 903, 917, 928, - 938, 950, 962, 975, 986, 1000, 1011 - ] - -# Updatable config blocks -updatable_config: - flagging: - kotekan_update_endpoint: "json" - start_time: 1535048997. - tag: "initial_flags" - bad_inputs: [ ] - gains: - kotekan_update_endpoint: "json" - start_time: 1541039597. - tag: "init_gain" - 26m_gated: - kotekan_update_endpoint: "json" - enabled: false - 26m_ungated: - kotekan_update_endpoint: "json" - enabled: false - -# Kotekan stages -buffer_recv: - n2: - kotekan_stage: bufferRecv - buf: visbuf_10s_all - listen_port: 11024 - cpu_affinity: [0, 8] - num_threads: 2 - 26m_ungated: - kotekan_stage: bufferRecv - buf: visbuf_5s_26m_ungated - listen_port: 11025 - 26m_gated: - kotekan_stage: bufferRecv - buf: visbuf_5s_26m_gated - listen_port: 11026 - -switch_26m_ungated: - kotekan_stage: bufferSwitch - in_bufs: - - enabled: visbuf_5s_26m_ungated - out_buf: visbuf_5s_26m_ungated_post - updatable_config: "/updatable_config/26m_ungated" - -switch_26m_gated: - kotekan_stage: bufferSwitch - in_bufs: - - enabled: visbuf_5s_26m_gated - out_buf: visbuf_5s_26m_gated_post - updatable_config: "/updatable_config/26m_gated" - -apply_flags: - kotekan_stage: receiveFlags - in_buf: visbuf_10s_all - out_buf: visbuf_10s_flags - updatable_config: "/updatable_config/flagging" - -vis_debug: - n2: - kotekan_stage: visDebug - in_buf: visbuf_10s_flags - 26m_ungated: - kotekan_stage: visDebug - in_buf: visbuf_5s_26m_ungated_post - 26m_gated: - kotekan_stage: visDebug - in_buf: visbuf_5s_26m_gated_post - -count_check: - n2: - kotekan_stage: countCheck - in_buf: visbuf_10s_flags - 26m_ungated: - kotekan_stage: countCheck - in_buf: visbuf_5s_26m_ungated - 26m_gated: - kotekan_stage: countCheck - in_buf: visbuf_5s_26m_gated - -apply_gains: - cpu_affinity: [4, 5] - num_threads: 2 - kotekan_stage: applyGains - in_buf: visbuf_10s_flags - out_buf: visbuf_10s_gain - gains_dir: "/test/data" - updatable_config: "/updatable_config/gains" - -stacking: - kotekan_stage: baselineCompression - in_buf: visbuf_10s_gain - out_buf: visbuf_10s_stack - stack_type: chime_in_cyl - exclude_inputs: [ - 46, 142, 688, 944, 960, 1058, 1166, 1225, 1314, 1521, 1543, 2032, 2034 - ] - num_threads: 2 - cpu_affinity: [2, 3] - -buffer_status: - kotekan_stage: bufferStatus - print_status: false - -# Generate the calibration stream -cal_subset: - kotekan_stage: prodSubset - in_buf: visbuf_10s_flags - out_buf: visbuf_10s_cal - prod_subset_type: autos - -timing_subset: - kotekan_stage: prodSubset - in_buf: visbuf_10s_flags - out_buf: visbuf_10s_timing - prod_subset_type: only_inputs - input_list: [46, 142, 688, 944, 960, 1058, 1166, 1314, 1543, 2032, 2034] - -# Generate the monitoring freq full N^2 stream -mfreq_subset: - n2: - kotekan_stage: freqSubset - in_buf: visbuf_10s_flags - out_buf: visbuf_10s_mfreq - subset_list: *mfreq - stack: - kotekan_stage: freqSubset - in_buf: visbuf_10s_stack - out_buf: visbuf_10s_stack_mfreq - subset_list: *mapfreq - -buffer_writers: - file_length: 256 - root_path: /mnt/recv1/buffer/ - write_ev: True - file_base: buffer - - cal: - kotekan_stage: visCalWriter - in_buf: visbuf_10s_cal - instrument_name: chimecal - file_length: 256 - dir_name: cal - - timing: - kotekan_stage: visCalWriter - in_buf: visbuf_10s_timing - instrument_name: chimetiming - dir_name: timing - -remove_ev: - - chimestack: - kotekan_stage: removeEv - in_buf: visbuf_10s_stack - out_buf: visbuf_10s_stack_ne - - timing: - kotekan_stage: removeEv - in_buf: visbuf_10s_timing - out_buf: visbuf_10s_timing_ne - - -archive_writers: - - file_length: 512 - file_type: raw - root_path: /data/untransposed/ - - n2_mf: - kotekan_stage: visWriter - in_buf: visbuf_10s_mfreq - instrument_name: chimeN2 - - chimestack: - kotekan_stage: visWriter - in_buf: visbuf_10s_stack_ne - instrument_name: chimestack - - 26m: - kotekan_stage: visWriter - in_buf: visbuf_5s_26m_ungated_post - instrument_name: chime26m - - 26mgated: - kotekan_stage: visWriter - in_buf: visbuf_5s_26m_gated_post - instrument_name: chime26mgated - - cal: - kotekan_stage: visWriter - in_buf: visbuf_10s_cal - instrument_name: chimecal - file_length: 256 - - timing: - kotekan_stage: visWriter - in_buf: visbuf_10s_timing_ne - instrument_name: chimetiming - - -# Transmit part of the stack to the recv1 for testing -buffer_send: - server_ip: 10.1.50.11 - stack_mfreq: - kotekan_stage: bufferSend - buf: visbuf_10s_stack_mfreq - server_port: 14096 - diff --git a/old_tests/simulate-chime/run_test.sh b/old_tests/simulate-chime/run_test.sh deleted file mode 100755 index ad74c983..00000000 --- a/old_tests/simulate-chime/run_test.sh +++ /dev/null @@ -1,43 +0,0 @@ -#!/bin/bash - -# Build kotekan and dependencies -echo "===== Checking highfive repository. =============" -if cd highfive; then git pull && cd ..; else git clone --single-branch --branch extensible-datasets https://github.com/jrs65/highfive.git; fi - -echo "===== Checking kotekan repository. ==============" -if cd kotekan; then git pull; else git clone --single-branch --branch rn/stdout https://github.com/kotekan/kotekan.git && cd kotekan; fi -git status -if cd build; then cd ..; else mkdir build; fi -cd .. - -echo "===== Building docker images. ===================" -export DOCKER_CLIENT_TIMEOUT=120 -export COMPOSE_HTTP_TIMEOUT=120 -docker-compose -f docker-compose.yaml build - -echo "===== Starting docker images. ===================" -docker-compose -f docker-compose.yaml up --scale gpu-cn01=10 -d - -echo "===== Starting fake GPS server. =================" -while : ; do ( echo -ne "HTTP/1.0 200 OK\r\nContent-Length: $(wc -c T_WAIT - assert t1 - t0 < T_WAIT + T_HOW_SLOW_IS_COCO - - for p in farm.ports: - assert farm.counters()[p][ENDPT_NAME] == 1 - assert ENDPT_NAME in response - for h in farm.hosts: - assert h in response[ENDPT_NAME] - assert "status" in response[ENDPT_NAME][h] - assert "reply" in response[ENDPT_NAME][h] - - assert response[ENDPT_NAME][h]["status"] == 200 - assert response[ENDPT_NAME][h]["reply"] == request - - -def test_timestamp(farm, runner): - response = runner.client(TS_ENDPT_NAME) - for p in farm.ports: - assert farm.counters()[p][TS_ENDPT_NAME] == 1 - assert TS_ENDPT_NAME in response - for h in farm.hosts: - assert h in response[TS_ENDPT_NAME] - assert "status" in response[TS_ENDPT_NAME][h] - assert "reply" in response[TS_ENDPT_NAME][h] - - assert response[TS_ENDPT_NAME][h]["status"] == 200 - - response = runner.client(GET_TS_ENDPT_NAME) - for p in farm.ports: - assert farm.counters()[p][GET_TS_ENDPT_NAME] == 1 - assert GET_TS_ENDPT_NAME in response - assert "state" in response - assert "timestamps" in response["state"] - assert "test" in response["state"]["timestamps"] - timestamp = response["state"]["timestamps"]["test"] - assert isinstance(timestamp, float) - # This timestamps should be fresh. Test that it's between 0 and 10s old. - assert time.time() - timestamp > 0 - assert time.time() - timestamp < 10 diff --git a/tests/coco_runner.py b/tests/coco_runner.py index c0a73ee3..b1d7a7da 100644 --- a/tests/coco_runner.py +++ b/tests/coco_runner.py @@ -25,12 +25,14 @@ def testN(coco_runner): coco_runner.client("callN") """ +import asyncio import json import multiprocessing import socket import threading from time import sleep +import aiohttp import fakeredis import pytest import yaml @@ -181,7 +183,7 @@ def add_config(self, **extra_config): # Merge in config self.config = config.merge_dict_tree(self.config, extra_config) - def add_targets(self, group, count): + def add_targets(self, group, count, callback=None): """Add `count` target rest servers to the group `group`. The targets are created using the rest_server test fixture. @@ -200,7 +202,7 @@ def add_targets(self, group, count): targets = [] for _ in range(count): server = self.rest_server() - server.accept_all() + server.accept_all(callback=callback) server.start() # Remember it so we can stop it later @@ -325,7 +327,73 @@ def start_daemon(self, *args): # even after it successfully fetches the port. sleep(0.2) - def client(self, *args, expect_failure=False, no_daemon=False, no_backend=False): + def call_endpoint( + self, + endpoint: str, + method: str = "GET", + data: dict | None = None, + query: str | None = None, + ): + """Call a daemon endpoint directly. + + i.e. without using the client. Starts the daemon if it isn't + already running. + + Parameters + ---------- + endpoint : str + Endpoint to call + method : str + HTTP method to use. The only values honoured are "GET" and "POST". + Other values are taken to mean "GET" (the default). + data : dict, optional + A dict of data to serialize to send to the daemon. + query : str, optional + URL query, if any (i.e. the part after '?' in the URL). Should be + URI-encoded. + + Returns + ------- + aiothhtp.ClientResponse + The response + Any: + The response body. If JSON-encoded, it will be decoded. + """ + + # Can't be called after stop() + if self._daemon_result: + raise RuntimeError("called after daemon stop.") + self.start_daemon() + + # Build the URL + url = f"http://127.0.0.1:{self._daemon_port}/{endpoint}" + if query: + url += "?" + query + + async def _get_response(method, url): + async with aiohttp.ClientSession() as session: + if method == "POST": + method = session.post + else: + method = session.get + + async with method(url, json=data) as response: + try: + body = await response.json() + except aiohttp.ContentTypeError: + body = await response.text() + return (response, body) + + return asyncio.run(_get_response(method, url)) + + def client( + self, + *args, + expect_failure=False, + decode=False, + no_daemon=False, + no_backend=False, + ): """Invoke the coco client. Parameters @@ -335,12 +403,22 @@ def client(self, *args, expect_failure=False, no_daemon=False, no_backend=False) expect_failure : bool Controls whether a non-zero (failure) or zero (success) exit code will cause an assertion failure + decode : bool + If set to True, adds "--json" to the client command-line and then, + if the client exits sucessfully, JSON decodes the client output + and returns that. no_daemon : bool, optional If True, don't start the daemon before running the client. If the daemon is already running, this won't stop it. no_backend : bool If True, don't set the COCO_BACKEND envar. + + Returns + ------- + If `decode` is True, returns a JSON-decoded dict of the client output, + if possible. If `decode` is False (the default), or JSON-decoding + fails, returns the `click.Result` from the CliRunner. """ import traceback @@ -373,9 +451,16 @@ def client(self, *args, expect_failure=False, no_daemon=False, no_backend=False) else {"COCO_BACKEND": f"127.0.0.1:{self._daemon_port}"} ) + if decode: + args = ("--json", *args) + # Invoke result = runner.invoke(entry, args=args) + # Reset the coco client after the test. This needs to be done + # because the CliRunner doesn't run "coco" in standalone mode. + entry._coco_init = False + # Show traceback if one was created if ( result.exit_code @@ -388,15 +473,21 @@ def client(self, *args, expect_failure=False, no_daemon=False, no_backend=False) print(result.output) if expect_failure: - assert result.exit_code != 0 + assert result.exit_code != 0, "Client failed to exit with error" assert type(result.exception) is SystemExit else: - assert result.exit_code == 0 + assert result.exit_code == 0, ( + f"Client exited with error: {result.exit_code}" + ) assert result.exception is None - # Reset the coco client after the test. This needs to be done - # because the CliRunner doesn't run "coco" in standalone mode. - entry._coco_init = False + # Attempt JSON decoding, if requested + if decode: + try: + result = json.loads(result.stdout) + except json.JSONDecodeError as e: + print(f"Failed to decode output: {e}") + pass # Return the result to the client return result diff --git a/tests/conftest.py b/tests/conftest.py index 6f6c5c35..b5e85012 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,5 @@ """Common test fixtures""" -import json import traceback import pytest @@ -89,10 +88,6 @@ def mock_comet(rest_server): def _register_state(route, body): """Pretend to be comet's /register-state endpoint.""" - - # Decode body - body = json.loads(body) - return {"result": "success", "request": "get_state", "hash": body["hash"]} # Create a mock broker diff --git a/tests/rest_server.py b/tests/rest_server.py index 8158eaf5..ef0e0952 100644 --- a/tests/rest_server.py +++ b/tests/rest_server.py @@ -35,6 +35,12 @@ def record_hit(self, code, body, response=None): path = self.path.split("?", 1)[0] path = path.split("#", 1)[0] + # Decode body, if needed + try: + body = body.decode() + except AttributeError: + pass + # Pass back up to the RestServer instance self.server.rest_server.record_hit( path, Hit(self.path, self.command, body, int(code), response) @@ -67,7 +73,10 @@ def send_json(self, body, response): self.end_headers() # Response goes in the body - self.wfile.write(response.encode()) + try: + self.wfile.write(response.encode()) + except BrokenPipeError: + pass return def handle_route(self): @@ -79,34 +88,43 @@ def handle_route(self): # Split the path itself from the query query_split = self.path.split("?", 1) path = query_split[0] + query = None if len(query_split) > 1: - body = "?" + query_split[1] + query = query_split[1] # Split the path from the fragment fragment_split = path.split("#", 1) path = fragment_split[0] + fragment = None if len(fragment_split) > 1: - if body: - body += "#" + fragment_split[1] - else: - body = "#" + fragment_split[1] + fragment = fragment_split[1] - # Read the body from a POST - if self.command == "POST": + # Read the body + if self.headers["Content-Length"]: body = self.rfile.read(int(self.headers["Content-Length"])) + if isinstance(body, bytes): + body = body.decode() + body = json.loads(body) else: - body = "" + body = {} - # If we're accepting all routes, reply with something generic + # Handle accepting any route if self.server.rest_server.any_route: - response = {} - if self.command == "POST": - # For a POST, send back what we were given - response["body"] = json.loads(body) - - # Add generic response data - response["path"] = path - response["result"] = "success" + if self.server.rest_server.any_callback: + response = self.server.rest_server.any_callback(path, body) + else: + # If no callback, generate a generic response + response = {} + if body: + response["body"] = body + if query: + response["query"] = query + if fragment: + response["fragment"] = fragment + + # Add generic response data + response["path"] = path + response["result"] = "success" # Return the response. self.send_json(body, response) @@ -118,7 +136,7 @@ def handle_route(self): if route.method == self.command: # Handle the route if route.callback: - response = route.callback(route, body.decode()) + response = route.callback(route, body) else: response = route.response # Return the response. @@ -130,7 +148,7 @@ def handle_route(self): error_code = ( HTTPStatus.METHOD_NOT_ALLOWED if bad_method else HTTPStatus.NOT_FOUND ) - self.record_hit(error_code, body) + self.record_hit(error_code, body if body else query) self.send_error(error_code) return @@ -164,6 +182,9 @@ def __init__(self): # If True, all routes are accepted self.any_route = False + # If accepting all routes, this will be called to create the return value + self.any_callback = None + # List of routes added with add_route self.routes = [] @@ -176,9 +197,10 @@ def __init__(self): # Is the server running? self.running = False - def accept_all(self): + def accept_all(self, callback=None): """Set up this rest server to accept any endpoint.""" self.any_route = True + self.any_callback = callback def add_route(self, path, method="GET", callback=None, response=None): """Add a route to the server. @@ -328,11 +350,8 @@ def _compare(a, b, equal, path=""): return None for hit in self.hits(route): - # decode the request - received = json.loads(hit.request.decode()) - # Return on first success - result = _compare(received, sent, full) + result = _compare(hit.request, sent, full) if not result: return @@ -340,5 +359,5 @@ def _compare(a, b, equal, path=""): pytest.fail( f'Data not received by route "{route}": {result}\n' f"Expected\n {sent}\nReceived:\n " - "\n ".join([str(hit.request.decode()) for hit in self.hits(route)]) + "\n ".join([str(hit.request) for hit in self.hits(route)]) ) diff --git a/tests/test_call_coco.py b/tests/test_call_coco.py new file mode 100644 index 00000000..795e3e9b --- /dev/null +++ b/tests/test_call_coco.py @@ -0,0 +1,52 @@ +"""Test call forwarding to other coco endpoints,""" + + +def test_forward(coco_runner): + """Test that endpoints forward.""" + + coco_runner.add_targets("test", 2) + coco_runner.add_endpoint( + "proxy", + { + "call": {"coco": ["end", "end2"]}, + "group": "test", + "values": {"foo": "int", "bar": "str"}, + }, + ) + coco_runner.add_endpoint( + "end", {"group": "test", "values": {"foo": "int", "bar": "str"}} + ) + coco_runner.add_endpoint( + "end2", {"group": "test", "values": {"foo": "int", "bar": "str"}} + ) + + result = coco_runner.client("end", "--foo=123", "--bar=abc", decode=True) + assert set(result) == {"end", "queue_wait", "success"} + assert result["success"] is True + assert result["end"]["200"] == 2 + + result = coco_runner.client("end2", "--foo=456", "--bar=def", decode=True) + assert set(result) == {"end2", "queue_wait", "success"} + assert result["success"] is True + assert result["end2"]["200"] == 2 + + # Now via proxy + result = coco_runner.client("proxy", "--foo=789", "--bar=ghi", decode=True) + assert set(result) == {"end", "end2", "proxy", "queue_wait", "success"} + assert result["success"] is True + + # Check endpoints + assert result["proxy"]["200"] == 2 + assert result["end"]["end"]["200"] == 2 + assert result["end"]["success"] is True + assert result["end2"]["end2"]["200"] == 2 + assert result["end2"]["success"] is True + + # Check hits on the targets + for target in coco_runner.targets: + assert target.hit_count("/end") == 2 + target.assert_hit_received("/end", {"foo": 123, "bar": "abc"}) + target.assert_hit_received("/end", {"foo": 789, "bar": "ghi"}) + assert target.hit_count("/end2") == 2 + target.assert_hit_received("/end2", {"foo": 456, "bar": "def"}) + target.assert_hit_received("/end2", {"foo": 789, "bar": "ghi"}) diff --git a/tests/test_checks.py b/tests/test_checks.py new file mode 100644 index 00000000..627e531e --- /dev/null +++ b/tests/test_checks.py @@ -0,0 +1,291 @@ +"""Test endpoint calls triggered by failures.""" + +import random +from threading import Lock + +fixed_num = 0 +lock = Lock() + + +def callback(path, body): + """Used by the rest_server to generate data + to send back to cocod.""" + + global fixed_num + + if path == "/rand": + # If the body has a True "rand" value, return + # a random number. Otherwise return a fixed number. + with lock: + if body["rand"]: + rand = random.random() + + fixed_num += rand + return {"rand": rand} + return {"rand": fixed_num} + + # Otherwise, just return body + return body + + +def test_forward(coco_runner): + """Test coco's forward reply check.""" + + # Number of targets + N_HOSTS = 2 + + # Create rest farm with the above callback + coco_runner.add_targets("test", N_HOSTS, callback=callback) + + # Add endpoints + coco_runner.add_endpoint( + "type_check", + { + "group": "test", + "values": {"ok": "bool"}, + "call": {"forward": {"name": "pong", "reply": {"type": {"ok": "bool"}}}}, + }, + ) + coco_runner.add_endpoint( + "type_check_fail", + { + "group": "test", + "values": {"ok": "int"}, + "call": {"forward": {"name": "pong", "reply": {"type": {"ok": "bool"}}}}, + }, + ) + coco_runner.add_endpoint( + "value_check", + { + "group": "test", + "values": {"ok": "bool"}, + "call": {"forward": {"name": "pong", "reply": {"value": {"ok": True}}}}, + }, + ) + coco_runner.add_endpoint( + "identical_check", + { + "group": "test", + "values": {"rand": "bool"}, + "call": {"forward": {"name": "rand", "reply": {"identical": ["rand"]}}}, + }, + ) + coco_runner.add_endpoint("pong", {"group": "test"}) + coco_runner.add_endpoint("rand", {"group": "test"}) + + # For before check + coco_runner.add_endpoint( + "bvalue_check", + { + "group": "test", + "values": {"ok": "bool"}, + "call": {"forward": None}, + "before": {"name": "pong", "reply": {"value": {"ok": True}}}, + }, + ) + coco_runner.add_endpoint( + "avalue_check", + { + "group": "test", + "values": {"ok": "bool"}, + "call": {"forward": None}, + "after": {"name": "pong", "reply": {"value": {"ok": True}}}, + }, + ) + coco_runner.add_endpoint( + "save_to_state", + { + "group": "test", + "values": {"a": "bool", "b": "bool"}, + "call": {"forward": None}, + "save_state": "fo/bar", + }, + ) + coco_runner.add_endpoint( + "save_to_state_fu", + { + "group": "test", + "values": {"b": "bool"}, + "call": {"forward": None}, + "save_state": "fu/bar", + }, + ) + coco_runner.add_endpoint( + "state_check_path", + { + "group": "test", + "values": {"b": "bool"}, + "call": {"forward": {"name": "pong", "reply": {"state": "fu/bar"}}}, + }, + ) + coco_runner.add_endpoint( + "state_check_values", + { + "group": "test", + "values": {"a": "bool", "b": "bool"}, + "call": { + "forward": { + "name": "pong", + "reply": {"state": {"a": "fo/bar/a", "b": "fo/bar/b"}}, + } + }, + }, + ) + + # Test failed type check + response = coco_runner.client( + "-r", "FULL", "type_check_fail", "--ok=1", decode=True + ) + for p in coco_runner.targets: + p.assert_hit_received("/pong", {"ok": 1}) + + # Check failure report + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"]) + assert len(failed_host) == N_HOSTS + reply = response["failed_checks"]["pong"][failed_host[0]]["reply"] + assert reply["type"] == ["ok"] + + # Test passing type check + response = coco_runner.client("type_check", "--no-ok", decode=True) + for p in coco_runner.targets: + assert p.hit_count("/pong") == 2 + + # Check failure report + assert response["success"] is True + assert "failed_checks" not in response + + # Test failed value check + response = coco_runner.client("value_check", "--no-ok", decode=True) + for p in coco_runner.targets: + assert p.hit_count("/pong") == 3 + assert response["success"] is False + + response = coco_runner.client("-r", "FULL", "value_check", "--no-ok", decode=True) + for p in coco_runner.targets: + assert p.hit_count("/pong") == 4 + + # Check failure report + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"]) + assert len(failed_host) == N_HOSTS + reply = response["failed_checks"]["pong"][failed_host[0]]["reply"] + assert reply["value"] == ["ok"] + + # Test passing value check + response = coco_runner.client("value_check", "--ok", decode=True) + for p in coco_runner.targets: + assert p.hit_count("/pong") == 5 + + # Check failure report + assert response["success"] is True + assert "failed_checks" not in response + + # Test failed identical check + response = coco_runner.client( + "-r", "FULL", "identical_check", "--rand", decode=True + ) + for p in coco_runner.targets: + assert p.hit_count("/rand") == 1 + + # Check failure report + assert response["success"] is False + failed_host = list(response["failed_checks"]["rand"].keys()) + assert len(failed_host) == N_HOSTS + reply = response["failed_checks"]["rand"][failed_host[0]]["reply"] + assert reply["not_identical"] == ["all"] + + # Test passing identical check + response = coco_runner.client("identical_check", "--no-rand", decode=True) + for p in coco_runner.targets: + assert p.hit_count("/rand") == 2 + + # Check failure report + assert response["success"] is True + assert "failed_checks" not in response + + # Test failed check on before + response = coco_runner.client("-r", "FULL", "bvalue_check", "--ok", decode=True) + for p in coco_runner.targets: + assert p.hit_count("/pong") == 6 + + # Check failure report + assert response["success"] is False + failed_host = list(response["pong"]["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS + reply = response["pong"]["failed_checks"]["pong"][failed_host[0]]["reply"] + assert reply["missing"] == ["ok"] + + # Test failed check on after + response = coco_runner.client("-r", "FULL", "avalue_check", "--ok", decode=True) + for p in coco_runner.targets: + assert p.hit_count("/pong") == 7 + + # Check failure report + assert response["success"] is False + failed_host = list(response["pong"]["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS + reply = response["pong"]["failed_checks"]["pong"][failed_host[0]]["reply"] + assert reply["missing"] == ["ok"] + + # Test state checks + # ------------------------------------------------------------------- + # First without setting it (will always fail) + response = coco_runner.client("-r", "FULL", "state_check_path", "-b", decode=True) + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS + + response = coco_runner.client( + "-r", "FULL", "state_check_path", "--no-b", decode=True + ) + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS + + # Set the state + response = coco_runner.client("save_to_state_fu", "--no-b", decode=True) + assert response["success"] is True + + # Test again + response = coco_runner.client("-r", "FULL", "state_check_path", "-b", decode=True) + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS + + response = coco_runner.client("state_check_path", "--no-b", decode=True) + assert response["success"] is True + assert "failed_checks" not in response + + # The same with multiple paths to compare between state and reply + # ------------------------------------------------------------------- + # First without setting it (will always fail) + response = coco_runner.client( + "-r", "FULL", "state_check_values", "-a", "-b", decode=True + ) + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS + + response = coco_runner.client( + "-r", "FULL", "state_check_values", "--no-a", "--no-b", decode=True + ) + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS + + # Set the state + response = coco_runner.client("save_to_state", "--no-a", "--no-b", decode=True) + assert response["success"] is True + + # Test again + response = coco_runner.client("state_check_values", "--no-a", "--no-b", decode=True) + assert response["success"] is True + assert "failed_checks" not in response + + response = coco_runner.client( + "-r", "FULL", "state_check_values", "--no-a", "-b", decode=True + ) + assert response["success"] is False + failed_host = list(response["failed_checks"]["pong"].keys()) + assert len(failed_host) == N_HOSTS diff --git a/tests/test_client.py b/tests/test_client.py index 0ccd6aea..1f863107 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -51,6 +51,7 @@ def test_style(coco_runner): expected_result = { "endpoint": f"http://127.0.0.1:{coco_runner.port}/nop", "method": "GET", + "data": {"coco_report_type": "CODES_OVERVIEW"}, } # Default is yaml @@ -91,6 +92,7 @@ def test_show_call(coco_runner): assert result == { "endpoint": f"http://127.0.0.1:{coco_runner.port}/nop", "method": "GET", + "data": {"coco_report_type": "CODES_OVERVIEW"}, } # A local endpoint @@ -101,8 +103,12 @@ def test_show_call(coco_runner): ) assert result == { "endpoint": f"http://127.0.0.1:{coco_runner.port}/update-blocklist", - "method": "POST", - "data": {"command": "add", "hosts": ["HOST"]}, + "method": "GET", + "data": { + "command": "add", + "hosts": ["HOST"], + "coco_report_type": "CODES_OVERVIEW", + }, } # coco config -- the endpoint here is not called, but --show-call-only @@ -314,5 +320,5 @@ def test_value_handling(coco_runner): assert target.hit_count("/endpoint") == 1 hit = target.hits("/endpoint")[0] assert hit.method == "POST" - assert json.loads(hit.request) == data + assert hit.request == data assert hit.response == {"path": "/endpoint", "result": "success", "body": data} diff --git a/tests/test_forward.py b/tests/test_forward.py new file mode 100644 index 00000000..f509517e --- /dev/null +++ b/tests/test_forward.py @@ -0,0 +1,93 @@ +"""Test basic endpoint call forwarding.""" + +import pytest + + +@pytest.fixture +def runner(coco_runner): + """coco_runner with pre-made endpoint. + + (And targets) + """ + + coco_runner.add_targets("test", 2) + coco_runner.add_endpoint( + "test", + { + "group": "test", + "report_latency": True, + "values": {"foo": "int", "bar": "str"}, + }, + ) + return coco_runner + + +def test_forward(runner): + """Test if a request gets forwarded to an external endpoint.""" + request = {"foo": 0, "bar": "1337"} + response = runner.client("-r", "FULL", "test", "--foo=0", "--bar=1337", decode=True) + + assert "test" in response + for t in runner.targets: + assert t.hit_count("/test") == 1 + h = f"http://127.0.0.1:{t.port}/" + assert h in response["test"] + assert "status" in response["test"][h] + assert "reply" in response["test"][h] + + assert response["test"][h]["status"] == 200 + assert response["test"][h]["reply"]["body"] == request + + runner.client("test", "--foo=0", "--bar=1337") + runner.client("test", "--foo=0", "--bar=1337") + for t in runner.targets: + assert t.hit_count("/test") == 3 + + +def test_wrong_vars(runner): + request = {"foo": "dfg", "bar": 1337} + response, _ = runner.call_endpoint("test", data=request) + assert response.status == 400 + + request = {"foo": 1337} + response, _ = runner.call_endpoint("test", data=request) + assert response.status == 400 + + +def test_url_args(runner): + """Test if URL arguments get forwarded to an external endpoint.""" + + request = {"foo": 0, "bar": "1337"} + request_full = request.copy() + request_full.update({"coco_report_type": "FULL"}) + params = {"cat": "1", "hat": "rat"} + query_str = "&".join([f"{k}={params[k]}" for k in params]) + + response, json = runner.call_endpoint("test", data=request_full, query=query_str) + assert response.status == 200 + + assert "test" in json + for t in runner.targets: + h = f"http://127.0.0.1:{t.port}/" + assert h in json["test"] + assert "status" in json["test"][h] + assert "reply" in json["test"][h] + assert "query" in json["test"][h]["reply"] + assert ( + "latency" in json["test"][h] + ) # Check that per-host latencies are returned + + assert json["test"][h]["status"] == 200 + assert json["test"][h]["reply"] == { + "body": request, + "path": "/test", + "query": query_str, + "result": "success", + } + + +def test_latency_stats(runner): + """Test if latency stats are returned from external endpoint""" + + response = runner.client("test", "--foo=0", "--bar=1337", decode=True) + assert "latency_stats" in response["test"] diff --git a/tests/test_on_failure.py b/tests/test_on_failure.py new file mode 100644 index 00000000..bcb99cf4 --- /dev/null +++ b/tests/test_on_failure.py @@ -0,0 +1,92 @@ +"""Test endpoint calls triggered by failures.""" + +from threading import Lock + +count = 0 +lock = Lock() + + +def callback(path, body): + """Reply with the incoming json request.""" + global count + + # for /restart, just return input + if path == "/restart": + return body + + # When count gets to 1, return an invalid reply + with lock: + if count > 0: + return {"not_ok": True} + count += 1 + return {"ok": True} + + +def test_on_reply(coco_runner): + """Test coco's on_failure option.""" + global count + + N_HOSTS = 2 + coco_runner.add_targets("test", N_HOSTS, callback=callback) + coco_runner.add_endpoint( + "call_single", + { + "group": "test", + "call": { + "forward": { + "name": "status", + "reply": {"type": {"ok": "bool"}}, + "on_failure": {"call_single_host": "restart"}, + } + }, + }, + ) + coco_runner.add_endpoint( + "call_all", + { + "group": "test", + "call": { + "forward": { + "name": "status", + "reply": {"type": {"ok": "bool"}}, + "on_failure": {"call": "restart"}, + } + }, + }, + ) + coco_runner.add_endpoint("status", {"group": "test"}) + coco_runner.add_endpoint("restart", {"group": "test"}) + + # Test call on failure + response = coco_runner.client("-r", "FULL", "call_all", decode=True) + for t in coco_runner.targets: + assert t.hit_count("/status") == 1 + assert t.hit_count("/restart") == 1 + + # Check failure report + failed_host = list(response["failed_checks"]["status"].keys()) + assert len(failed_host) == 1 + reply = response["failed_checks"]["status"][failed_host[0]]["reply"] + assert reply["missing"] == ["ok"] + + # reset count + with lock: + count = 0 + + # Test call_single_host + response = coco_runner.client("-r", "FULL", "call_single", decode=True) + + # Check failure report + failed_host = list(response["failed_checks"]["status"].keys()) + assert len(failed_host) == 1 + reply = response["failed_checks"]["status"][failed_host[0]]["reply"] + assert reply["missing"] == ["ok"] + + # Check only failed host called restart + for t in coco_runner.targets: + assert t.hit_count("/status") == 2 + # only second host should have failed + if t.port == int(failed_host[0].strip("http://").split(":")[1]): + assert t.hit_count("/restart") == 2 + else: + assert t.hit_count("/restart") == 1 diff --git a/tests/test_save_to_state.py b/tests/test_save_to_state.py new file mode 100644 index 00000000..c2d854e0 --- /dev/null +++ b/tests/test_save_to_state.py @@ -0,0 +1,81 @@ +"""Test endpoint config option `save_state` and `get_state`.""" + +import pytest + + +@pytest.fixture +def runner(coco_runner): + """Create a coco runner with some endpoints.""" + + coco_runner.add_endpoint( + "save", + { + "call": {"forward": None}, + "save_state": ["test_state/1", "test_state/2"], + "values": {"val": "int"}, + }, + ) + coco_runner.add_endpoint( + "get1", {"call": {"forward": None}, "get_state": "test_state/1"} + ) + coco_runner.add_endpoint( + "get2", {"call": {"forward": None}, "get_state": "test_state/2"} + ) + + return coco_runner + + +def test_save_state(runner): + """Test get/save_state.""" + # State starts off empty + + response = runner.client("get1", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "1" in response["state"]["test_state"] + assert response["state"]["test_state"]["1"] == {} + response = runner.client("get2", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "2" in response["state"]["test_state"] + assert response["state"]["test_state"]["2"] == {} + + runner.client("save", "--val=5") + response = runner.client("get1", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "1" in response["state"]["test_state"] + assert response["state"]["test_state"]["1"] == {"val": 5} + response = runner.client("get2", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "2" in response["state"]["test_state"] + assert response["state"]["test_state"]["2"] == {"val": 5} + + +def test_no_reset(runner): + """Check state with it initially set.""" + runner.set_state({"test_state": {"1": {"val": 5}, "2": {"val": 5}}}) + response = runner.client("get1", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "1" in response["state"]["test_state"] + assert response["state"]["test_state"]["1"] == {"val": 5} + response = runner.client("get2", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "2" in response["state"]["test_state"] + assert response["state"]["test_state"]["2"] == {"val": 5} + + +def test_reset(runner): + response = runner.client("get1", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "1" in response["state"]["test_state"] + assert response["state"]["test_state"]["1"] == {} + response = runner.client("get2", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert "2" in response["state"]["test_state"] + assert response["state"]["test_state"]["2"] == {} diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py new file mode 100644 index 00000000..9b853d3c --- /dev/null +++ b/tests/test_scheduler.py @@ -0,0 +1,86 @@ +"""Test endpoint scheduler.""" + +import time + +import yaml + + +def test_sched(coco_runner): + """Test if scheduled endpoints are called when they should be.""" + PERIOD = 0.5 + + coco_runner.add_targets("test", 2, callback=lambda path, data: data) + coco_runner.add_endpoint( + "scheduled", {"group": "test", "schedule": {"period": PERIOD}} + ) + coco_runner.add_endpoint( + "scheduled-check-type", + { + "group": "test", + "schedule": { + "period": PERIOD, + "require_state": {"path": "test/success", "type": "bool"}, + }, + }, + ) + coco_runner.add_endpoint( + "scheduled-check-val", + { + "group": "test", + "schedule": { + "period": PERIOD, + "require_state": { + "path": "test/success", + "type": "bool", + "value": True, + }, + }, + }, + ) + coco_runner.add_endpoint( + "scheduled-fail-type", + { + "group": "test", + "schedule": { + "period": PERIOD, + "require_state": {"path": "test/fail_type", "type": "bool"}, + }, + }, + ) + coco_runner.add_endpoint( + "scheduled-fail-val", + { + "group": "test", + "schedule": { + "period": PERIOD, + "require_state": { + "path": "test/fail_val", + "type": "bool", + "value": True, + }, + }, + }, + ) + + # This is not a state file, but we'll throw it into the storage_path, just because + # it's a convenient place to put it. + with open(coco_runner.storage_path / "TEST.yaml", "w") as f: + yaml.dump({"success": True, "fail_type": "not_a_bool", "fail_val": False}, f) + coco_runner.add_config( + load_state={"test": str(coco_runner.storage_path / "TEST.yaml")} + ) + + coco_runner.start_daemon() + # Let at least three periods pass + time.sleep(3.5 * PERIOD) + coco_runner.stop() + + for t in coco_runner.targets: + # We don't know this precisely, because more periods may + # happen during the coco_runner daemon set-up and teardown + assert t.hit_count("/scheduled") >= 3 + num_sched = t.hit_count("/scheduled") + assert t.hit_count("/scheduled-check-type") == num_sched + assert t.hit_count("/scheduled-check-val") == num_sched + assert t.hit_count("/scheduled-fail-type") == 0 + assert t.hit_count("/scheduled-fail-val") == 0 diff --git a/tests/test_set_state.py b/tests/test_set_state.py new file mode 100644 index 00000000..4f03537e --- /dev/null +++ b/tests/test_set_state.py @@ -0,0 +1,224 @@ +"""Test endpoint config option `set_state` and `get_state`.""" + +import pytest + +from coco.util import hash_dict + + +@pytest.fixture +def runner(coco_runner): + """A configured coco_runner.""" + + coco_runner.add_targets("test", 1, callback=lambda path, data: data) + coco_runner.add_config(exclude_from_reset=["excluded/from/reset"]) + coco_runner.add_endpoint("getall", {"call": {"forward": None}, "get_state": "/"}) + coco_runner.add_endpoint( + "set", {"call": {"forward": None}, "set_state": {"test_state": True}} + ) + coco_runner.add_endpoint( + "set_int", {"call": {"forward": None}, "set_state": {"test_state": 5}} + ) + coco_runner.add_endpoint( + "set_dict", + { + "call": {"forward": None}, + "set_state": {"test_state": {"s": {"n": {"a": "fu"}}}}, + }, + ) + coco_runner.add_endpoint( + "set_excluded_state", + { + "call": {"forward": None}, + "set_state": {"excluded/from/reset": 5}, + }, + ) + coco_runner.add_endpoint( + "get", {"call": {"forward": None}, "get_state": "test_state"} + ) + coco_runner.add_endpoint( + "get_excluded_state", + {"call": {"forward": None}, "get_state": "excluded/from/reset"}, + ) + coco_runner.add_endpoint( + "check_hash", + { + "group": "test", + "values": {"data": "str"}, + "call": { + "forward": {"name": "hash", "reply": {"state_hash": {"data": "/"}}} + }, + }, + ) + coco_runner.add_endpoint( + "check_hash2", + { + "group": "test", + "values": {"data": "str"}, + "call": { + "forward": { + "name": "hash", + "reply": {"state_hash": {"data": "test_state/s/n"}}, + } + }, + }, + ) + coco_runner.add_endpoint( + "check_state", + { + "group": "test", + "values": {"data": "str"}, + "call": { + "forward": { + "name": "hash", + "reply": {"state": {"data": "test_state/s/n/a"}}, + } + }, + }, + ) + coco_runner.add_endpoint( + "check_state2", + { + "group": "test", + "values": {"n": "dict"}, + "call": {"forward": {"name": "hash", "reply": {"state": "test_state/s/"}}}, + }, + ) + + return coco_runner + + +def test_get_state(runner): + """Test get/set_state.""" + + # Set state to True + runner.client("set") + + # Get state and compare + response = runner.client("get", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert response["state"]["test_state"] is True + + # Set state to INT + runner.client("set_int") + + # Get state and compare + response = runner.client("get", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert response["state"]["test_state"] == 5 + + # Test passing check against state hash + response = runner.client( + "-r", "FULL", "check_hash", "--data", hash_dict({"test_state": 5}), decode=True + ) + assert "failed_checks" not in response + + # Test failing check against state hash + response = runner.client( + "-r", "FULL", "check_hash", "--data", hash_dict({"foo": 5}), decode=True + ) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state_hash": ["data"]}} + + response = runner.client( + "-r", "FULL", "check_hash", "--data", hash_dict({"test_state": 4}), decode=True + ) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state_hash": ["data"]}} + + # The same for a part of the state: + runner.client("set_dict") + # Test passing check against partly state hash + response = runner.client( + "check_hash2", "--data", hash_dict({"a": "fu"}), decode=True + ) + assert "failed_checks" not in response + + # Test failing check against partly state hash + response = runner.client( + "-r", "FULL", "check_hash2", "--data", hash_dict({"a": "foo"}), decode=True + ) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state_hash": ["data"]}} + + # Test checks against state + # Test passing check against part of state + response = runner.client("check_state", "--data=fu", decode=True) + assert "failed_checks" not in response + + # Test failing check against part of state + response = runner.client("-r", "FULL", "check_state", "--data=f00", decode=True) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state": ["data"]}} + + response = runner.client("-r", "FULL", "check_state", "--data=", decode=True) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state": ["data"]}} + + # Test passing check against state + response = runner.client("check_state2", "-n", '{"a": "fu"}', decode=True) + assert "failed_checks" not in response + + # Test failing check against state + response = runner.client( + "-r", "FULL", "check_state2", "-n", '{"n": {"a": 0}}', decode=True + ) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state": ["all"]}} + + response = runner.client( + "-r", "FULL", "check_state2", "-n", '{"aa": "fu"}', decode=True + ) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state": ["all"]}} + + response = runner.client("-r", "FULL", "check_state2", "-n", "{}", decode=True) + assert "failed_checks" in response + assert "hash" in response["failed_checks"] + for r in response["failed_checks"]["hash"].values(): + assert r == {"reply": {"mismatch_with_state": ["all"]}} + + +def test_reset_state(runner): + runner.client("set_int") + response = runner.client("get", decode=True) + assert "state" in response + assert "test_state" in response["state"] + assert response["state"]["test_state"] == 5 + + runner.client("set_excluded_state") + response = runner.client("get_excluded_state", decode=True) + assert "state" in response + r = response["state"] + for p in ("excluded", "from", "reset"): + assert p in r + r = r[p] + assert r == 5 + + runner.client("reset-state") + response = runner.client("get", decode=True) + assert "status_code" in response + assert response["status_code"] == 500 # path not found + + response = runner.client("get_excluded_state", decode=True) + assert "state" in response + r = response["state"] + for p in ("excluded", "from", "reset"): + assert p in r + r = r[p] + assert r == 5 diff --git a/tests/test_timeout.py b/tests/test_timeout.py new file mode 100644 index 00000000..37dba5b6 --- /dev/null +++ b/tests/test_timeout.py @@ -0,0 +1,36 @@ +"""Test basic endpoint call forwarding with timeout.""" + +import time + + +def callback(path, data): + """Reply with the incoming json request.""" + time.sleep(2) + return data + + +def test_timeout_for_specific_forward(coco_runner): + """Test endpoint timeout""" + + coco_runner.add_targets("test", 2, callback=callback) + coco_runner.add_endpoint( + "test", + { + "call": {"forward": {"name": "test", "timeout": "1s"}}, + "group": "test", + "values": {"foo": "int", "bar": "str"}, + }, + ) + + response = coco_runner.client( + "-r", "FULL", "test", "--foo=0", "--bar=1337", decode=True + ) + + assert "test" in response + for t in coco_runner.targets: + h = f"http://127.0.0.1:{t.port}/" + assert h in response["test"] + assert "status" in response["test"][h] + assert "reply" in response["test"][h] + assert response["test"][h]["status"] == 0 + assert response["test"][h]["reply"] == "Timeout" diff --git a/tests/test_timeout_global.py b/tests/test_timeout_global.py new file mode 100644 index 00000000..fffb4146 --- /dev/null +++ b/tests/test_timeout_global.py @@ -0,0 +1,37 @@ +"""Test basic endpoint call forwarding with timeout.""" + +import time + + +def callback(path, data): + """Reply with the incoming json request.""" + time.sleep(2) + return data + + +def test_timeout_global(coco_runner): + """Test global timeout""" + + coco_runner.add_config(timeout="1s") + coco_runner.add_targets("test", 2, callback=callback) + coco_runner.add_endpoint( + "test", + { + "call": {"forward": {"name": "test"}}, + "group": "test", + "values": {"foo": "int", "bar": "str"}, + }, + ) + + response = coco_runner.client( + "-r", "FULL", "test", "--foo=0", "--bar=1337", decode=True + ) + + assert "test" in response + for t in coco_runner.targets: + h = f"http://127.0.0.1:{t.port}/" + assert h in response["test"] + assert "status" in response["test"][h] + assert "reply" in response["test"][h] + assert response["test"][h]["status"] == 0 + assert response["test"][h]["reply"] == "Timeout" diff --git a/tests/test_wait.py b/tests/test_wait.py new file mode 100644 index 00000000..236b3aeb --- /dev/null +++ b/tests/test_wait.py @@ -0,0 +1,65 @@ +"""Test the internal WAIT endpoint. Also test timestamps.""" + +import time + + +def test_wait(coco_runner): + """Test the wait endpoint.""" + + coco_runner.add_targets("test", 2) + coco_runner.add_endpoint( + "test", + { + "group": "test", + "values": {"foo": "int", "bar": "str"}, + "call": {"coco": {"name": "wait", "request": {"duration": "2s"}}}, + }, + ) + + t0 = time.monotonic() + response = coco_runner.client( + "-r", "FULL", "test", "--foo=0", "--bar=1337", decode=True + ) + t1 = time.monotonic() + assert t1 - t0 > 2 + assert t1 - t0 < 5 + + assert "test" in response + for t in coco_runner.targets: + assert t.hit_count("/test") == 1 + h = f"http://127.0.0.1:{t.port}/" + assert h in response["test"] + assert "status" in response["test"][h] + assert response["test"][h]["status"] == 200 + + +def test_timestamp(coco_runner): + """Test timestamps""" + + coco_runner.add_targets("test", 2, callback=lambda path, data: data) + coco_runner.add_endpoint( + "ts_endpt", {"group": "test", "timestamp": "timestamp/test"} + ) + coco_runner.add_endpoint( + "get_ts_endpt", {"group": "test", "get_state": "timestamp/test"} + ) + + response = coco_runner.client("ts_endpt", decode=True) + assert "ts_endpt" in response + for t in coco_runner.targets: + assert t.hit_count("/ts_endpt") == 1 + + response = coco_runner.client("-r", "FULL", "get_ts_endpt", decode=True) + for t in coco_runner.targets: + assert t.hit_count("/get_ts_endpt") == 1 + + assert "get_ts_endpt" in response + assert "state" in response + assert "timestamp" in response["state"] + assert "test" in response["state"]["timestamp"] + timestamp = response["state"]["timestamp"]["test"] + assert isinstance(timestamp, float) + + # This timestamps should be fresh. Test that it's between 0 and 10s old. + assert time.time() - timestamp > 0 + assert time.time() - timestamp < 10