gofactroy代码工厂生成python,go

''' @Title 文件名称: AttDaysController_test.gotest @Description 描述: 测试APIAttDaysController @Author 作者: leijianming 时间(2024-03-31 16:03:58) @Update 作者: leijianming 时间(2024-03-31 16:03:58) ''' import json, logging, unittest,datetime from django.db import transaction from base.dto import IcHandler from urllib import request from base.dto.IcHandler import IcHandler import json from urllib import error from django.test import TestCase from django.test import TransactionTestCase from base.consts.consts import _const import requests, os, django, kunlong from datetime import datetime #from base.utils import KunlongUtils base_url = _const.base_url_local #'http://127.0.0.1:8000/v1' base_url_entity = base_url + "/attdays" os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_tagify.settings") # project_name 项目名称 os.environ.setdefault("INSTALLED_APPS", "kunlong") # project_name 项目名称 django.setup() from kunlong.compute.model.attdaysmodel import AttDaysModel from kunlong.compute.service.attdaysservice import AttDaysService class TestAttDaysController(unittest.TestCase): # TransactionTestCase TestCase会自动删除数据 logger = logging.getLogger(__name__) # class TestDefault(TestCase): # TransactionTestCase TestCase @classmethod def setUpClass(cls): pass #os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_tagify.settings") #os.environ.setdefault("INSTALLED_APPS", "kunlong") # django.setup() @classmethod def tearDownClass(cls): pass def tearDown(self) -> None: pass def setUp(self): pass # 设置配置 # settings.IGNORE_WECHAT_SIGNATURE = True #os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_tagify.settings") #os.environ.setdefault("INSTALLED_APPS", "kunlong") #django.setup() # settings.configure() # @transaction.atomic # def post(self, request): # # 设置事务保存点 # s1 = transaction.savepoint() # 可以设置多个保存点 # # # 数据库操作。。。 # # # 事务回滚 (如果发生异常,就回滚事务) # transaction.savepoint_rollback(s1) # 可以回滚到指定的保存点 # # # 提交事务 (如果没有异常,就提交事务) # transaction.savepoint_commit(s1) def insert(self): entity = AttDaysModel() #entity.createdAt = datetime.now() entity.ini(False) # KunlongUtils.transDatetime2Str(entity) entity.transDatetime2Str() #entity.dirName = "etstDirrname" entity.save() # entity = IcDirModel.objects.create(dir_name="补助计算", tenant_id="租户1111", active=True, user_id=0, # update_user_id=0, parent_dir_id=0, depth=1) # result = IcDirModel.objects.filter(dir_id=entity.dir_id).values() # transaction.savepoint_commit(savepoint1) # print(IcHandler.IcHandler.dumps2json(result[0])) return entity.daysId #@transaction.atomic def test0001_insert(self): id = self.insert() print(id) def buildQueryParam(self): dict_data = {} dict_data['orderBys'] = "-daysId" dict_data['start'] = 0 dict_data['limit'] = 3 param = {} param['daysId'] = None dict_data['param'] = param return dict_data def test0002_query_service(self): # def test0002_query_service(self): id = self.insert() #body = '''{"orderBys":"-daysId","start":0,"limit":3,"param":{"daysId":id}} ''' result = AttDaysService.query(**self.buildQueryParam()) print(result.dumps2json()) def test0003_create(self): entity = AttDaysModel() # entity.createdAt = datetime.now() entity.ini(False) entity.transDatetime2Str() dict = entity.__dict__ del dict['baseModel'] del dict['_state'] e = AttDaysModel.objects.create(**dict) print(str(e)) #用POST方法向Restful API发生消息 def test0004_save(self): try: create_url = base_url_entity + '/save' # the Restful api url create_headers = {'cookie': 'the cookie'} # the request headers body_data = ''' { "daysId": null, "active": true, "createdAt": "2021-1-1", "depth": 1, "dirName": "补助2计算", "parentDirId": 0, "tenantId": "租户1lll", "updatedAt": "2011-12-21 21:01:11", "updateUserId": 0, "userId": 0 } ''' entity = AttDaysModel() #entity.createdAt = datetime.now() entity.ini(False) #KunlongUtils.transDatetime2Str(entity) entity.transDatetime2Str() # entity.dirName="etstDirrname" body_data = bytes(str(entity), 'utf8') resp = requests.post(create_url, headers=create_headers, data=body_data) result_dict = json.loads(resp.text) #result = json.dumps(obj=result_dict, indent=4, ensure_ascii=False) print(IcHandler.dumps2jsonPretty(result_dict)) # print(result_dict['data']) r = self.findById(result_dict['data']) print(IcHandler.dumps2jsonPretty(r)) except error.HTTPError as err: error_body = err.file.read().decode() print(json.loads(error_body)) def findById(self, id): query_url_addr = base_url_entity + '/findById?id=' + str(id) # the Restful api url query_headers = {'cookie': 'the cookie'} # the request headers resp = requests.get(query_url_addr, headers=query_headers) result_json = json.loads(resp.text) # the json object of response data return result_json # json.dumps(obj=result_json, indent=4, ensure_ascii=False) def test0005_findById(self): id = self.insert() r = self.findById(id) #print(type(r)) print(IcHandler.dumps2jsonPretty(r)) def test0006_findByIds(self): values = {} values['id1'] = self.insert() values['id2'] = self.insert() url = base_url_entity + '/findByIds?ids={id1},{id2}' # the Restful api url print("Start findByIds: " + url.format(**values)) create_headers = {'cookie': 'the cookie'} # the request headers response = requests.get(url.format(**values), headers=create_headers) result_json = json.loads(response.text) print(IcHandler.dumps2jsonPretty(result_json)) print("End findByIds: " + url.format(**values)) #{"orderBys":"-dir_id","start":0,"limit":3,"param":{"dir_id":null}} #用POST方法向Restful API发生消息 def test0007_query(self): id = self.insert() print(id) try: url = base_url_entity + '/query' # the Restful api url create_headers = {'cookie': 'the cookie'} # the request headers #body = '''{"orderBys":"-daysId","start":0,"limit":3,"param":{"daysId":null}} ''' body_data = bytes(IcHandler.dumps2jsonPretty(self.buildQueryParam()), 'utf8') req = request.Request(url, headers=create_headers, data=body_data, method='POST') resp = request.urlopen(req) result = resp.read().decode() result_json = json.loads(result) #result = json.dumps(obj=result_json, indent=4, ensure_ascii=False) print(IcHandler.dumps2jsonPretty(result_json)) except error.HTTPError as err: error_body = err.file.read().decode() #print(json.loads(result)) def test0008_query(self): id = self.insert() print(id) try: url = base_url_entity