diff --git a/.idea/workspace.xml b/.idea/workspace.xml index 4b07338..5a681fb 100644 --- a/.idea/workspace.xml +++ b/.idea/workspace.xml @@ -4,288 +4,12 @@ - @@ -436,6 +167,7 @@ - \ No newline at end of file diff --git a/unit-backend/config/settings.py b/unit-backend/config/settings.py index 50fb237..87b64eb 100644 --- a/unit-backend/config/settings.py +++ b/unit-backend/config/settings.py @@ -328,3 +328,7 @@ } ASGI_APPLICATION = "config.asgi.application" + +CONSUL_HOST = "127.0.0.1" +CONSUL_PORT = 8500 +CONSUL_DNS_PORT = 8600 diff --git a/unit-backend/utils/grpconsul.py b/unit-backend/utils/grpconsul.py new file mode 100644 index 0000000..9ea4eae --- /dev/null +++ b/unit-backend/utils/grpconsul.py @@ -0,0 +1,49 @@ +import random +import socket +from dns import resolver +from config import settings + + +class ServiceConsul(object): + + def __init__(self): + self.consul_host = settings.CONSUL_HOST + self.consul_dns_port = settings.CONSUL_DNS_PORT + + @staticmethod + def extract_ip_prefix(encoded_address: str): + match = encoded_address.split(".")[0] + return match + + def decode_address(self, encoded_address): + hex_address = self.extract_ip_prefix(encoded_address) + binary_address = bytes.fromhex(hex_address) + return socket.inet_ntoa(binary_address) + + def fetch_user_service_addresses(self, srv_record_name): + custom_resolver = resolver.Resolver() + custom_resolver.nameservers = [self.consul_host] + custom_resolver.port = self.consul_dns_port + + try: + answer = custom_resolver.resolve(srv_record_name, 'SRV') + selected_rdata = random.choice(answer) + ip_address = self.decode_address(selected_rdata.target.to_text()) + return ip_address, selected_rdata.port + except resolver.NoAnswer as e: + print(f"No answer found for the query: {srv_record_name}") + except resolver.NXDOMAIN as e: + print(f"Domain not found: {srv_record_name}") + except Exception as e: + print(f"Error occurred while resolving DNS: {str(e)}") + + +def main(): + srv_record_name = 'unit_executor.service.consul' + cons = ServiceConsul() + address, port = cons.fetch_user_service_addresses(srv_record_name) + print(address, port) + + +if __name__ == "__main__": + main() diff --git a/unit-executor/register/consul.py b/unit-executor/register/consul.py index 26f2b2a..4a99869 100644 --- a/unit-executor/register/consul.py +++ b/unit-executor/register/consul.py @@ -1,8 +1,4 @@ -import uuid -from abc import ABC import consul -import requests -import random class ConsulRegister(object): diff --git a/unit-executor/requirements.txt b/unit-executor/requirements.txt index 68c2793..21d08cf 100644 --- a/unit-executor/requirements.txt +++ b/unit-executor/requirements.txt @@ -1,4 +1,4 @@ -atomic_bomb_engine==0.41.3 +atomic_bomb_engine~=0.41.0 Faker==8.11.0 grpcio==1.64.1 jmespath==1.0.1 @@ -7,10 +7,14 @@ loguru==0.6.0 MySQL-python==1.2.5 numpy==1.24.2 protobuf==5.28.0 -pydantic==2.8.2 +pydantic~=2.7.2 python_dateutil==2.8.2 Requests==2.32.3 requests_toolbelt==1.0.0 rsa==4.7.2 simplejson==3.17.0 -urllib3==2.2.2 +urllib3~=2.2.1 +typing~=3.7.4.3 +python-dateutil~=2.8.2 +asyncio~=3.4.3 +mysqlclient~=2.1.1 \ No newline at end of file diff --git a/unit-executor/unitrunner/test/base.py b/unit-executor/unitrunner/test/base.py index 3b7e0ef..16a27b5 100644 --- a/unit-executor/unitrunner/test/base.py +++ b/unit-executor/unitrunner/test/base.py @@ -1,66 +1,115 @@ -import json -import sys -from unitrunner.engine.base import run_test, run_api -from utils.logger import logger +# import json +# import sys +# from unitrunner.engine.base import run_test, run_api +# from utils.logger import logger +# +# if __name__ == '__main__': +# # "loop/for/while/http/if" +# case_data = [{ +# 'name': "测试场景名称1", +# 'cases': [{ +# "title": "测试用例2", +# "host": "http://httpbin.org/post", +# "interface": { +# "url": "/post", +# "name": "登录", +# "method": "post", +# }, +# "headers": { +# 'content-Type': "application/json" +# }, +# "request": { +# 'json': {"mobile_phone": "${{user_mobile}}", "pwd": "lemonban"}, +# }, +# 'setup_script': "print('前置脚本123')", +# 'teardown_script': "test.assertion('相等',200,response.status_code)" +# }] +# }] +# config = { +# 'ENV': { +# "host": 'http://httpbin.org', +# 'user_mobile': 999999999}, +# 'db': [{}, {}], +# 'global_func': "print('前置脚本123')", +# 'rerun': 1 +# } +# # response = run_test(env_config=config, case_data=case_data, debug=False) +# # sys.stdout.write("测试结果\n") +# # sys.stdout.write(str(response)) +# # sys.stdout.write("\n测试结果\n") +# api_doc = api_data = { +# "type": "loop", +# "parameters": { +# "count": 2 +# }, +# "children": [{ +# "type": "http", +# "title": "demo", +# "interface": { +# "url": "http://httpbin.org/post", +# "name": "33333", +# "method": "POST" +# }, +# "headers": {}, +# "request": {"data": {}}, +# "setup_script": "", +# "teardown_script": "", +# "extract": {}, +# "validators": [] +# }] +# } +# responses = run_api(api_data=api_doc) +# sys.stdout.write(str(responses)) +# logger.info( +# f"-------- 测试结果 ----------\n" +# f"{json.dumps(responses, indent=4, ensure_ascii=False)}\n" +# ) -if __name__ == '__main__': - # "loop/for/while/http/if" - case_data = [{ - 'name': "测试场景名称1", - 'cases': [{ - "title": "测试用例2", - "host": "http://httpbin.org/post", - "interface": { - "url": "/post", - "name": "登录", - "method": "post", - }, - "headers": { - 'content-Type': "application/json" - }, - "request": { - 'json': {"mobile_phone": "${{user_mobile}}", "pwd": "lemonban"}, - }, - 'setup_script': "print('前置脚本123')", - 'teardown_script': "test.assertion('相等',200,response.status_code)" - }] - }] - config = { - 'ENV': { - "host": 'http://httpbin.org', - 'user_mobile': 999999999}, - 'db': [{}, {}], - 'global_func': "print('前置脚本123')", - 'rerun': 1 - } - # response = run_test(env_config=config, case_data=case_data, debug=False) - # sys.stdout.write("测试结果\n") - # sys.stdout.write(str(response)) - # sys.stdout.write("\n测试结果\n") - api_doc = api_data = { - "type": "loop", - "parameters": { - "count": 2 - }, - "children": [{ - "type": "http", - "title": "demo", - "interface": { - "url": "http://httpbin.org/post", - "name": "33333", - "method": "POST" - }, - "headers": {}, - "request": {"data": {}}, - "setup_script": "", - "teardown_script": "", - "extract": {}, - "validators": [] - }] - } - responses = run_api(api_data=api_doc) - sys.stdout.write(str(responses)) - logger.info( - f"-------- 测试结果 ----------\n" - f"{json.dumps(responses, indent=4, ensure_ascii=False)}\n" - ) +import random +import re +import socket +import struct +from dns import resolver + +# 配置自定义 DNS 服务器 +custom_resolver = resolver.Resolver() +custom_resolver.nameservers = ['127.0.0.1'] +# 设置端口号（这里假设所有服务器都使用 8600 端口） +custom_resolver.port = 8600 # 注意：通常不需要对每个服务器单独设置，除非它们使用不同端口 + + +def extract_ip_prefix(encoded_address: str): + match = encoded_address.split(".")[0] + return match + + +def decode_address(encoded_address): + hex_address = extract_ip_prefix(encoded_address) + binary_address = bytes.fromhex(hex_address) + return socket.inet_ntoa(binary_address) + + +def srv_to_address_random_choice(srv_record_name): + try: + answer = custom_resolver.resolve(srv_record_name, 'SRV') + selected_rdata = random.choice(answer) + ip_address = decode_address(selected_rdata.target.to_text()) + print(f"Service: {srv_record_name} - {ip_address}:{selected_rdata.port} ") + return ip_address, selected_rdata.port + except resolver.NoAnswer as e: + print(f"No answer found for the query: {srv_record_name}") + except resolver.NXDOMAIN as e: + print(f"Domain not found: {srv_record_name}") + except Exception as e: + print(f"Error occurred while resolving DNS: {str(e)}") + + +def main(): + # 查询 SRV 记录 + srv_record_name = 'unit_executor.service.consul' + # srv_record_name = 'consul.service.consul' + srv_to_address_random_choice(srv_record_name) + + +if __name__ == "__main__": + main()