zwoop commented on code in PR #12648: URL: https://github.com/apache/trafficserver/pull/12648#discussion_r2511212226
########## tests/gold_tests/autest-site/ats_replay.test.ext: ########## @@ -0,0 +1,144 @@ +''' Review Comment: This is fine for now, but at some point, we should add a section to our dev "testing" documentation, on how and when people should use this new framework. ########## tests/gold_tests/autest-site/ats_replay.test.ext: ########## @@ -0,0 +1,144 @@ +''' +Implement general-purpose ATS test extensions using proxy verifier replay files. +''' +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from typing import Optional +import os +import yaml + + +def configure_ats(obj: 'TestRun', server: 'Process', ats_config: dict, dns: Optional['Process'] = None): + '''Configure ATS per the configuration in the replay file. + + :param obj: The Test object to configure ATS for. + :param server: The Proxy Verifier Server process to use. + :param ats_config: The ATS configuration from the replay file. + :param dns: The DNS process to use. Optional. + :returns: The ATS process object. + ''' + name = ats_config.get('name', 'ts') + process_config = ats_config.get('process_config', {}) + ts = obj.MakeATSProcess(name, **process_config) + records_config = ats_config.get('records_config', {}) + ts.Disk.records_config.update(records_config) + remap_config = ats_config.get('remap_config', []) + for remap_entry in remap_config: + if isinstance(remap_entry, str): + ts.Disk.remap_config.AddLine(remap_entry) + elif isinstance(remap_entry, dict): + from_url = remap_entry['from'] + to_url = remap_entry['to'] + to_url = to_url.replace('{SERVER_HTTP_PORT}', str(server.Variables.http_port)) + to_url = to_url.replace('{SERVER_HTTPS_PORT}', str(server.Variables.https_port)) + plugins = remap_entry.get('plugins', []) + line = f'map {from_url} {to_url}' + for plugin in plugins: + line += f' @plugin={plugin["name"]}' + for arg in plugin["args"]: + line += f' @pparam={arg}' + ts.Disk.remap_config.AddLine(line) + if dns: + ts.Disk.records_config.update( + { + 'proxy.config.dns.nameservers': f'127.0.0.1:{dns.Variables.Port}', + 'proxy.config.dns.resolv_conf': 'NULL', + }) + + for item in ats_config.get('copy_to_config_dir', []): + item_path = os.path.join(obj.TestDirectory, item) + if os.path.isdir(item_path): + src_dir = item_path + dst_dir = os.path.join(ts.Variables.CONFIGDIR, item) + ts.Setup.MakeDir(dst_dir) + ts.Setup.Copy(src_dir, dst_dir) + else: + ts.Setup.CopyAs(item, ts.Variables.CONFIGDIR) + return ts + + +def ATSReplayTest(obj, replay_file: str): + '''Create a TestRun that configures ATS and runs HTTP traffic using the replay file. + + :param obj: The Test object to add the test run to. + :param replay_file: Replay file specifying the test configuraiton and test traffic. + :returns: The TestRun object. + ''' + + replay_path = replay_file if os.path.isabs(replay_file) else os.path.join(obj.TestDirectory, replay_file) + with open(replay_path, 'r') as f: + replay_config = yaml.load(f, Loader=yaml.FullLoader) + + # The user must specify the 'autest' node. + if not 'autest' in replay_config: + raise ValueError(f"Replay file {replay_file} does not contain 'autest' section") + autest_config = replay_config['autest'] + + tr = obj.AddTestRun(autest_config['description']) + + # Copy the specified files and directories down. + tr.Setup.Copy(replay_file, tr.RunDirectory) + + for files_to_copy in autest_config.get('files_to_copy', []): + tr.Setup.Copy(files_to_copy) + for dirs_to_copy in autest_config.get('dirs_to_copy', []): + tr.Setup.Copy(dirs_to_copy) + + # DNS configuration. Review Comment: For safety, should we add a ``` dns = None ``` here ? ########## tests/gold_tests/autest-site/ats_replay.test.ext: ########## @@ -0,0 +1,144 @@ +''' +Implement general-purpose ATS test extensions using proxy verifier replay files. +''' +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from typing import Optional +import os +import yaml + + +def configure_ats(obj: 'TestRun', server: 'Process', ats_config: dict, dns: Optional['Process'] = None): + '''Configure ATS per the configuration in the replay file. + + :param obj: The Test object to configure ATS for. + :param server: The Proxy Verifier Server process to use. + :param ats_config: The ATS configuration from the replay file. + :param dns: The DNS process to use. Optional. + :returns: The ATS process object. + ''' + name = ats_config.get('name', 'ts') + process_config = ats_config.get('process_config', {}) + ts = obj.MakeATSProcess(name, **process_config) + records_config = ats_config.get('records_config', {}) + ts.Disk.records_config.update(records_config) + remap_config = ats_config.get('remap_config', []) + for remap_entry in remap_config: + if isinstance(remap_entry, str): + ts.Disk.remap_config.AddLine(remap_entry) + elif isinstance(remap_entry, dict): + from_url = remap_entry['from'] + to_url = remap_entry['to'] + to_url = to_url.replace('{SERVER_HTTP_PORT}', str(server.Variables.http_port)) + to_url = to_url.replace('{SERVER_HTTPS_PORT}', str(server.Variables.https_port)) + plugins = remap_entry.get('plugins', []) + line = f'map {from_url} {to_url}' + for plugin in plugins: + line += f' @plugin={plugin["name"]}' + for arg in plugin["args"]: + line += f' @pparam={arg}' + ts.Disk.remap_config.AddLine(line) + if dns: + ts.Disk.records_config.update( + { + 'proxy.config.dns.nameservers': f'127.0.0.1:{dns.Variables.Port}', + 'proxy.config.dns.resolv_conf': 'NULL', + }) + + for item in ats_config.get('copy_to_config_dir', []): + item_path = os.path.join(obj.TestDirectory, item) + if os.path.isdir(item_path): + src_dir = item_path + dst_dir = os.path.join(ts.Variables.CONFIGDIR, item) + ts.Setup.MakeDir(dst_dir) + ts.Setup.Copy(src_dir, dst_dir) + else: + ts.Setup.CopyAs(item, ts.Variables.CONFIGDIR) + return ts + + +def ATSReplayTest(obj, replay_file: str): + '''Create a TestRun that configures ATS and runs HTTP traffic using the replay file. + + :param obj: The Test object to add the test run to. + :param replay_file: Replay file specifying the test configuraiton and test traffic. Review Comment: Typo ########## tests/gold_tests/autest-site/ats_replay.test.ext: ########## @@ -0,0 +1,144 @@ +''' +Implement general-purpose ATS test extensions using proxy verifier replay files. +''' +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from typing import Optional +import os +import yaml + + +def configure_ats(obj: 'TestRun', server: 'Process', ats_config: dict, dns: Optional['Process'] = None): + '''Configure ATS per the configuration in the replay file. + + :param obj: The Test object to configure ATS for. + :param server: The Proxy Verifier Server process to use. + :param ats_config: The ATS configuration from the replay file. + :param dns: The DNS process to use. Optional. + :returns: The ATS process object. + ''' + name = ats_config.get('name', 'ts') + process_config = ats_config.get('process_config', {}) + ts = obj.MakeATSProcess(name, **process_config) + records_config = ats_config.get('records_config', {}) + ts.Disk.records_config.update(records_config) + remap_config = ats_config.get('remap_config', []) + for remap_entry in remap_config: + if isinstance(remap_entry, str): + ts.Disk.remap_config.AddLine(remap_entry) + elif isinstance(remap_entry, dict): + from_url = remap_entry['from'] + to_url = remap_entry['to'] + to_url = to_url.replace('{SERVER_HTTP_PORT}', str(server.Variables.http_port)) + to_url = to_url.replace('{SERVER_HTTPS_PORT}', str(server.Variables.https_port)) + plugins = remap_entry.get('plugins', []) + line = f'map {from_url} {to_url}' + for plugin in plugins: + line += f' @plugin={plugin["name"]}' + for arg in plugin["args"]: + line += f' @pparam={arg}' + ts.Disk.remap_config.AddLine(line) + if dns: + ts.Disk.records_config.update( + { + 'proxy.config.dns.nameservers': f'127.0.0.1:{dns.Variables.Port}', + 'proxy.config.dns.resolv_conf': 'NULL', + }) + + for item in ats_config.get('copy_to_config_dir', []): + item_path = os.path.join(obj.TestDirectory, item) + if os.path.isdir(item_path): + src_dir = item_path + dst_dir = os.path.join(ts.Variables.CONFIGDIR, item) + ts.Setup.MakeDir(dst_dir) + ts.Setup.Copy(src_dir, dst_dir) + else: + ts.Setup.CopyAs(item, ts.Variables.CONFIGDIR) + return ts + + +def ATSReplayTest(obj, replay_file: str): + '''Create a TestRun that configures ATS and runs HTTP traffic using the replay file. + + :param obj: The Test object to add the test run to. + :param replay_file: Replay file specifying the test configuraiton and test traffic. + :returns: The TestRun object. + ''' + + replay_path = replay_file if os.path.isabs(replay_file) else os.path.join(obj.TestDirectory, replay_file) + with open(replay_path, 'r') as f: + replay_config = yaml.load(f, Loader=yaml.FullLoader) Review Comment: The LLM suggest using yaml.SafeLoader instead here: ``` `FullLoader` is deprecated in recent PyYAML versions in favor of `SafeLoader` ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
