evyos-center-server/wag-managment-api-service-version-2
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

alchemy flush and save functions updated

Browse Source
This commit is contained in:
berkay
2024-11-11 18:55:53 +03:00
parent c42a19c262
commit 1f1222c32d
163 changed files with 6296 additions and 476 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
service_app_test/test_application/evyos/address_building.py
View File

@@ -1,11 +1,12 @@
from random import randint
from faker import Faker
from test_application.evyos.bases import requester, active_and_confirmed
from test_application.evyos.datas.company_employee_data import list_options
from test_application.evyos.datas.get_building_types import list_building_types
from test_application.evyos.datas.get_occupants_codes import get_occupants_types
from test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
from service_app_test.test_application.evyos.datas.company_employee_data import list_options
from service_app_test.test_application.evyos.datas.get_building_types import list_building_types
from service_app_test.test_application.evyos.datas.get_occupants_codes import get_occupants_types
from service_app_test.test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
from service_app_test.bases import active_and_confirmed
fake = Faker()
@@ -55,7 +56,7 @@ build_living_space = lambda build_part_uu_id, life_person_uu_id, occupant_type_u
}
def create_build_parts(build_uu_id, count: int = 1):
def create_build_parts(requester, build_uu_id, count: int = 1):
flat_type = list_building_types(type_code="DAIRE")
flat_type_uu_id = flat_type["data"][0]["uu_id"]
directions = get_type_codes_key_and_class(class_name="Directions", key_name="NN")
@@ -76,7 +77,7 @@ def create_build_parts(build_uu_id, count: int = 1):
# return response.json()
def create_post_code(post_code):
def create_post_code(post_code, requester):
print("post_code", post_code)
post_code_response = requester.post(
endpoint="/postcode/create",
@@ -87,7 +88,7 @@ def create_post_code(post_code):
return
def create_addresses(address):
def create_addresses(address, requester):
address_response = requester.post(
endpoint="/address/create",
data=address,
@@ -97,7 +98,7 @@ def create_addresses(address):
return
def create_build_with_address_uu_id(address_dict_with_uu_id):
def create_build_with_address_uu_id(address_dict_with_uu_id, requester):
response = requester.post(
endpoint="/building/build/create",
data={**address_dict_with_uu_id, "build_number": "11"},
@@ -107,7 +108,7 @@ def create_build_with_address_uu_id(address_dict_with_uu_id):
return response.json()
def create_build_living_space(living_space_dict):
def create_build_living_space(living_space_dict, requester):
response = requester.post(
endpoint="/building/living_space/create",
data=living_space_dict,
@@ -117,7 +118,7 @@ def create_build_living_space(living_space_dict):
return response.json()
def search_street(search_text):
def search_street(search_text, requester):
response = requester.post(
endpoint="/address/search",
data={
@@ -141,7 +142,7 @@ def search_street(search_text):
return street_data["AddressStreet.uu_id"]
def list_post_codes():
def list_post_codes(requester):
post_code_response = requester.post(
endpoint="/postcode/list",
data=list_options,
@@ -151,7 +152,7 @@ def list_post_codes():
return post_code_response.json()
def list_addresses():
def list_addresses(requester):
address_response = requester.post(
endpoint="/address/list",
data=list_options,
@@ -161,7 +162,7 @@ def list_addresses():
return address_response.json()
def list_building(query=None):
def list_building(requester, query=None):
query_options = list_options
if query:
query_options["query"] = query
@@ -171,7 +172,7 @@ def list_building(query=None):
return response.json()
def list_build_living_space():
def list_build_living_space(requester):
response = requester.post(
endpoint="/building/living_space/list",
data=list_options,
@@ -181,7 +182,7 @@ def list_build_living_space():
return response.json()
def list_building_parts(query: dict = None):
def list_building_parts(requester,query: dict = None):
query_dict = query
if not query:
query_dict = {"part_code__contains": ":"}
@@ -196,64 +197,69 @@ def list_building_parts(query: dict = None):
def create_property_owner(
build_part_uu_id, person_uu_id, occupant_type_flat_owner_uu_id
build_part_uu_id, person_uu_id, occupant_type_flat_owner_uu_id, requester
):
return create_build_living_space(
living_space_dict=build_living_space(
build_part_uu_id=build_part_uu_id,
life_person_uu_id=person_uu_id,
occupant_type_uu_id=occupant_type_flat_owner_uu_id,
)
),
requester=requester,
)
def create_property_tenant(
build_part_uu_id, person_uu_id, occupant_type_flat_tenant_uu_id
build_part_uu_id, person_uu_id, occupant_type_flat_tenant_uu_id, requester
):
return create_build_living_space(
living_space_dict=build_living_space(
build_part_uu_id=build_part_uu_id,
life_person_uu_id=person_uu_id,
occupant_type_uu_id=occupant_type_flat_tenant_uu_id,
)
),
requester=requester,
)
def create_property_resident(
build_part_uu_id, person_uu_id, occupant_type_flat_resident_uu_id
build_part_uu_id, person_uu_id, occupant_type_flat_resident_uu_id, requester
):
return create_build_living_space(
living_space_dict=build_living_space(
build_part_uu_id=build_part_uu_id,
life_person_uu_id=person_uu_id,
occupant_type_uu_id=occupant_type_flat_resident_uu_id,
)
),
requester=requester,
)
def create_property_observer(build_part_uu_id, person_uu_id):
def create_property_observer(build_part_uu_id, person_uu_id, requester):
return create_build_living_space(
living_space_dict=build_living_space(
build_part_uu_id=build_part_uu_id,
life_person_uu_id=person_uu_id,
occupant_type_uu_id=None,
)
),
requester=requester,
)
def create_occupant_build_manager(
build_part_uu_id, person_uu_id, occupant_type_flat_manager_uu_id
build_part_uu_id, person_uu_id, occupant_type_flat_manager_uu_id, requester
):
return create_build_living_space(
living_space_dict=build_living_space(
build_part_uu_id=build_part_uu_id,
life_person_uu_id=person_uu_id,
occupant_type_uu_id=occupant_type_flat_manager_uu_id,
)
),
requester=requester,
)
def get_random_people_list():
def get_random_people_list(requester):
# person_uu_id : 2d5dac99-c445-44fc-a0a5-de97ec65781a is a tenant and lives in the build part as PL and PT
random_user_list_response = requester.post(
data={
@@ -277,7 +283,7 @@ def get_random_people_list():
def assign_random_people_to_build_parts(
list_of_building_parts, list_of_random_people, occupant_dict, build_uu_id
list_of_building_parts, list_of_random_people, occupant_dict, build_uu_id, requester
):
part_counter = 0
@@ -286,7 +292,7 @@ def assign_random_people_to_build_parts(
uu_id_occupant_type_flat_manager = occupant_dict["occupant_type_flat_manager"]
uu_id_occupant_type_flat_resident = occupant_dict["occupant_type_flat_resident"]
man_room = list_building_parts(query={"part_no": "0", "build_uu_id": build_uu_id})
man_room = list_building_parts(query={"part_no": "0", "build_uu_id": build_uu_id}, requester=requester)
man_room_uu_id = man_room["data"][0]["uu_id"]
for i in range(len(list_of_random_people)):
@@ -301,6 +307,7 @@ def assign_random_people_to_build_parts(
build_part_uu_id=selected_build_part_uu_id,
person_uu_id=list_of_random_people[i - 1]["uu_id"],
occupant_type_flat_owner_uu_id=uu_id_occupant_type_flat_owner,
requester=requester
)
# Property Tenant
@@ -308,6 +315,7 @@ def assign_random_people_to_build_parts(
build_part_uu_id=selected_build_part_uu_id,
person_uu_id=list_of_random_people[i]["uu_id"],
occupant_type_flat_tenant_uu_id=uu_id_occupant_type_flat_tenant,
requester=requester
)
# Property Tenant Living Space Occupant
@@ -315,6 +323,7 @@ def assign_random_people_to_build_parts(
build_part_uu_id=selected_build_part_uu_id,
person_uu_id=list_of_random_people[i]["uu_id"],
occupant_type_flat_resident_uu_id=uu_id_occupant_type_flat_resident,
requester=requester
)
if i == 1:
@@ -322,11 +331,12 @@ def assign_random_people_to_build_parts(
build_part_uu_id=man_room_uu_id,
person_uu_id=list_of_random_people[i]["uu_id"],
occupant_type_flat_manager_uu_id=uu_id_occupant_type_flat_manager,
requester=requester
)
def run_address_to_building():
street_uu_id = search_street("Reşat Nuri")
def run_address_to_building(requester):
street_uu_id = search_street("Reşat Nuri", requester=requester)
occupant_type_flat_owner = get_occupants_types(occupant_code="FL-OWN")
occupant_type_flat_tenant = get_occupants_types(occupant_code="FL-TEN")
@@ -340,11 +350,11 @@ def run_address_to_building():
occupant_type_flat_resident=occupant_type_flat_resident["data"]["uu_id"],
)
create_post_code(post_code=post_code_dict(street_uu_id))
selected_post_code_uu_id = list_post_codes()["data"][0]["uu_id"]
create_post_code(post_code=post_code_dict(street_uu_id), requester=requester)
selected_post_code_uu_id = list_post_codes(requester=requester)["data"][0]["uu_id"]
create_addresses(address=address_dict(selected_post_code_uu_id))
selected_address_uu_id = list_addresses()["data"][0]["uu_id"]
create_addresses(address=address_dict(selected_post_code_uu_id), requester=requester)
selected_address_uu_id = list_addresses(requester=requester)["data"][0]["uu_id"]
build_type_response = list_building_types(type_code="apt")
build_type_uu_id_ = build_type_response["data"][0]["uu_id"]
@@ -352,14 +362,15 @@ def run_address_to_building():
build_dict(
address_uu_id=selected_address_uu_id,
build_type_uu_id=build_type_uu_id_,
)
),
requester=requester
)
selected_build_uu_id = list_building()["data"][0]["uu_id"]
create_build_parts(selected_build_uu_id, count=21)
selected_build_uu_id = list_building(requester=requester)["data"][0]["uu_id"]
create_build_parts(build_uu_id=selected_build_uu_id, count=21, requester=requester)
random_people_list = get_random_people_list()
building_parts_list = list_building_parts()["data"]
random_people_list = get_random_people_list(requester=requester)
building_parts_list = list_building_parts(requester=requester)["data"]
print("building_parts_list", len(building_parts_list))
assign_random_people_to_build_parts(
@@ -367,5 +378,6 @@ def run_address_to_building():
list_of_building_parts=building_parts_list,
occupant_dict=occupant_type_dict,
build_uu_id=selected_build_uu_id,
requester=requester
)
print("list_build_living_space", list_build_living_space())
print("list_build_living_space", list_build_living_space(requester=requester))

144
service_app_test/test_application/evyos/bases.py
View File

@@ -1,144 +0,0 @@
import requests
from api_configs import Auth
active_and_confirmed = dict(
created_by="System",
confirmed_by="System",
is_confirmed=True,
active=True,
deleted=False,
is_notification_send=True,
)
class RequestToApi:
def __init__(self, base_url=None):
self.base_url = base_url or "http://0.0.0.0:41575/"
self.headers = {
"domain": "evyos.com.tr",
"Content-Type": "application/json",
"Accept": "application/json",
}
def set_access_token(self, access_token):
self.headers[Auth.ACCESS_TOKEN_TAG] = access_token
def url(self, endpoint):
return self.base_url + endpoint
def post(self, data, endpoint):
response = requests.post(self.url(endpoint), headers=self.headers, json=data)
return response
def get(self, endpoint):
print("url", self.url(endpoint))
response = requests.get(self.url(endpoint), headers=self.headers)
return response
def patch(self, data, endpoint):
response = requests.patch(self.url(endpoint), headers=self.headers, json=data)
return response
class FilterObject:
def __init__(
self,
page=None,
size=None,
order_field=None,
order_type=None,
include_joins=None,
query=None,
):
self.page = page or 1
self.size = size or 10
self.order_field = order_field or "id"
self.order_type = order_type or "asc"
self.include_joins = include_joins or []
self.query = query or {}
def dump(self):
return dict(
page=self.page,
size=self.size,
order_field=self.order_field,
order_type=self.order_type,
include_joins=list(self.include_joins),
query=dict(self.query),
)
def login_via_email_and_password(
login_data, is_password_valid=True, selection_list: list = None
):
if not is_password_valid:
if not login_data.get("password_token"):
raise Exception("Password token is not found")
change_password_with_token(password_token=login_data.get("password_token"))
login_dict = requester.post(endpoint="authentication/login", data=login_data)
print("login_dict", login_dict.text)
login_dict = login_dict.json()
content = login_dict
access_token = content.get("access_token")
refresh_token = content.get("refresh_token")
access_object = content.get("access_object")
if not access_object:
raise Exception("Access object is not found")
companies_uu_id_list = None
if access_object.get("user_type") == "employee":
companies_uu_id_list = access_object.get("companies_uu_id_list")
elif access_object.get("user_type") == "occupant":
companies_uu_id_list = access_object.get("available_occupants")
# Todo Create Company and assign bulk to Duty2Company
if not selection_list and not access_object.get("user_type") == "employee":
print("Please select a company and an occupant")
print(companies_uu_id_list)
exit()
requester.set_access_token(access_token)
if access_object.get("user_type") == "employee":
uu_id = companies_uu_id_list[0]
refresh_company = requester.post(
endpoint="authentication/select", data={"company_uu_id": uu_id}
)
return uu_id
elif access_object.get("user_type") == "occupant":
occupants = companies_uu_id_list[str(selection_list[0])]
occupant_uu_id = None
for occupant in occupants:
if occupant == str(selection_list[1]):
occupant_uu_id = occupant
break
return_dict = {
"occupant_uu_id": occupant_uu_id,
"build_part_uu_id": str(selection_list[0]),
}
refresh_company = requester.post(
endpoint="authentication/select", data=return_dict
)
return return_dict
def change_password_with_token(password_token):
data_dict = {
"password_token": password_token,
"password": "string",
"re_password": "string",
}
if password_token:
response = requester.post(
data=data_dict, endpoint="authentication/create_password"
)
print(response.json())
requester = RequestToApi()

41
service_app_test/test_application/evyos/company_employee.py
View File

@@ -1,24 +1,21 @@
from test_application.evyos.bases import (
requester,
# select_company_uu_id,
active_and_confirmed,
)
from test_application.evyos.people import (
from service_app_test.test_application.evyos.people import (
create_random_person,
create_random_user,
create_random_occupant_user,
)
from test_application.evyos.datas.company_employee_data import (
from service_app_test.test_application.evyos.datas.company_employee_data import (
departments,
duties,
staffs,
)
from service_app_test.bases import active_and_confirmed
print("Program up and running **-------------------------------------")
def create_department_via_list(list_of_departments):
def create_department_via_list(list_of_departments, requester):
for department in list_of_departments:
request_department = requester.post(
data={
@@ -30,7 +27,7 @@ def create_department_via_list(list_of_departments):
print(request_department.json())
def create_duty_via_list(list_of_duties):
def create_duty_via_list(list_of_duties, requester):
for duty in list_of_duties:
duty_dict = {
"duty_name": duty["duty_name"],
@@ -42,7 +39,7 @@ def create_duty_via_list(list_of_duties):
print(request_duty.json())
def create_duties_via_list(list_of_duties, company_uu_id):
def create_duties_via_list(list_of_duties, company_uu_id, requester):
for duty in list_of_duties:
department_response = requester.post(
endpoint="department/list",
@@ -68,7 +65,7 @@ def create_duties_via_list(list_of_duties, company_uu_id):
)
def create_staff_via_list(list_of_staff):
def create_staff_via_list(list_of_staff, requester):
for staff in list_of_staff:
duty_response = requester.post(
endpoint="duty/list",
@@ -89,7 +86,7 @@ def create_staff_via_list(list_of_staff):
print(request_staff.json())
def create_employee_via_list(list_of_staff):
def create_employee_via_list(list_of_staff, requester):
for staff in list_of_staff:
duty_response = requester.post(
endpoint="duty/list",
@@ -142,7 +139,7 @@ def create_employee_via_list(list_of_staff):
print(request_staff.json())
def create_people_via_list(people_count: int):
def create_people_via_list(people_count: int, requester):
for people in range(people_count):
people = create_random_person()
people_dict = people.model_dump()
@@ -159,7 +156,7 @@ def create_people_via_list(people_count: int):
selected_user_uu_id = response_user["data"]["uu_id"]
def create_occupant_people_via_list(people_count: int):
def create_occupant_people_via_list(people_count: int, requester):
for people in range(people_count):
people = create_random_person()
people_dict = people.model_dump()
@@ -176,14 +173,14 @@ def create_occupant_people_via_list(people_count: int):
selected_user_uu_id = response_user["data"]["uu_id"]
def run_company_and_depends(company_uu_id):
create_department_via_list(list_of_departments=departments)
create_duty_via_list(list_of_duties=duties)
create_duties_via_list(list_of_duties=duties, company_uu_id=company_uu_id)
create_staff_via_list(list_of_staff=staffs)
create_employee_via_list(list_of_staff=duties)
create_people_via_list(people_count=10)
create_occupant_people_via_list(people_count=70)
def run_company_and_depends(company_uu_id, requester):
create_department_via_list(list_of_departments=departments, requester=requester)
create_duty_via_list(list_of_duties=duties, requester=requester)
create_duties_via_list(list_of_duties=duties, company_uu_id=company_uu_id, requester=requester)
create_staff_via_list(list_of_staff=staffs, requester=requester)
create_employee_via_list(list_of_staff=duties, requester=requester)
create_people_via_list(people_count=10, requester=requester)
create_occupant_people_via_list(people_count=70, requester=requester)
print("Program ended **-------------------------------------")

6
service_app_test/test_application/evyos/datas/get_building_types.py
View File

@@ -1,8 +1,8 @@
from test_application.evyos.bases import requester
from test_application.evyos.datas.company_employee_data import list_options
def list_building_types(lang: str = "TR", type_code: str = None):
def list_building_types(requester, lang: str = "TR", type_code: str = None):
from service_app_test.test_application.evyos.datas.company_employee_data import list_options
options_smart_query = {
**list_options,
"query": {

5
service_app_test/test_application/evyos/datas/get_occupants_codes.py
View File

@@ -1,7 +1,6 @@
from test_application.evyos.bases import requester
def get_occupants_types(occupant_code):
def get_occupants_types(occupant_code, requester):
response = requester.post(
endpoint="/occupant_types/get/code",
data={"type_code": occupant_code},
@@ -11,7 +10,7 @@ def get_occupants_types(occupant_code):
return response.json()
def get_occupants_types_uu_id(occupant_uu_id):
def get_occupants_types_uu_id(occupant_uu_id, requester):
response = requester.post(
endpoint="/occupant_types/get/uu_id",
data={"uu_id": occupant_uu_id},

9
service_app_test/test_application/evyos/datas/get_type_codes.py
View File

@@ -1,7 +1,6 @@
from test_application.evyos.bases import requester
def get_type_codes_key_and_class(class_name, key_name):
def get_type_codes_key_and_class(class_name, key_name, requester):
response = requester.post(
endpoint="/enums/get/key",
data={"class_name": class_name, "key_name": key_name},
@@ -11,7 +10,7 @@ def get_type_codes_key_and_class(class_name, key_name):
return response.json()
def get_type_codes_via_uuid(uu_id):
def get_type_codes_via_uuid(uu_id, requester):
response = requester.post(
endpoint="/enums/get/uu_id",
data={"uu_id": uu_id},
@@ -21,7 +20,7 @@ def get_type_codes_via_uuid(uu_id):
return response.json()
def get_all_type_codes():
def get_all_type_codes(requester):
response = requester.post(
endpoint="/enums/list/all",
data={},
@@ -31,7 +30,7 @@ def get_all_type_codes():
return response.json()
def get_all_type_codes_with_class(class_name):
def get_all_type_codes_with_class(class_name, requester):
response = requester.post(
endpoint="/enums/list/class",
data={"class_name": class_name},

35
service_app_test/test_application/evyos/decision_book.py
View File

@@ -1,9 +1,9 @@
import arrow
from test_application.evyos.bases import requester, active_and_confirmed
from test_application.evyos.datas.company_employee_data import list_options
from test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
from test_application.evyos.address_building import list_building
from service_app_test.bases import active_and_confirmed
from service_app_test.test_application.evyos.datas.company_employee_data import list_options
from service_app_test.test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
from service_app_test.test_application.evyos.address_building import list_building
decision_book_dict = lambda build_uu_id, resp_company_uu_id: {
@@ -20,14 +20,14 @@ event_bind_to_occupant = lambda event_uu_id_list, build_living_space_uu_id: {
}
def get_uu_id_of_enum(enum_class: str, key: str):
def get_uu_id_of_enum(enum_class: str, key: str, requester):
get_type_codes_response = get_type_codes_key_and_class(
class_name=enum_class, key_name=key
class_name=enum_class, key_name=key, requester=requester
)
return get_type_codes_response["data"]["uu_id"]
def list_company():
def list_company(requester):
response = requester.post(
endpoint="/company/list",
data=list_options,
@@ -37,7 +37,7 @@ def list_company():
return response.json()
def list_decision_books():
def list_decision_books(requester):
response = requester.post(
endpoint="/build/decision_book/list",
data=list_options,
@@ -47,7 +47,7 @@ def list_decision_books():
return response.json()
def list_decision_book_items():
def list_decision_book_items(requester):
response = requester.post(
endpoint="/build/decision_book/items/list",
data=list_options,
@@ -57,7 +57,7 @@ def list_decision_book_items():
return response.json()
def create_decision_book(decision_book):
def create_decision_book(decision_book, requester):
response = requester.post(
endpoint="/build/decision_book/create",
data=decision_book,
@@ -67,7 +67,7 @@ def create_decision_book(decision_book):
return response.json()
def give_service_access_to_occupant(occupant_dict):
def give_service_access_to_occupant(occupant_dict, requester):
response = requester.post(
endpoint="/bind/services/occupant",
data=occupant_dict,
@@ -77,7 +77,7 @@ def give_service_access_to_occupant(occupant_dict):
return response.json()
def give_event_access_to_build_decision_book_items(occupant_dict):
def give_event_access_to_build_decision_book_items(occupant_dict, requester):
response = requester.post(
endpoint="/bind/events/occupant",
data={
@@ -92,18 +92,19 @@ def give_event_access_to_build_decision_book_items(occupant_dict):
return response.json()
def run_decision_book_depends():
build_response = list_building()
def run_decision_book_depends(requester):
build_response = list_building(requester=requester)
selected_build_uu_id = build_response["data"][0]["uu_id"]
company_response = list_company()
company_response = list_company(requester=requester)
selected_company_uu_id = company_response["data"][0]["uu_id"]
create_decision_book(
decision_book_dict(
build_uu_id=selected_build_uu_id,
resp_company_uu_id=selected_company_uu_id,
)
),
requester=requester,
)
decision_book = list_decision_books()["data"][0]["uu_id"]
decision_book = list_decision_books(requester=requester)["data"][0]["uu_id"]
return decision_book

23
service_app_test/test_application/evyos/decision_book_items.py
View File

@@ -1,6 +1,6 @@
from test_application.evyos.bases import requester, active_and_confirmed
from test_application.evyos.datas.get_occupants_codes import get_occupants_types
from test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
from service_app_test.bases import active_and_confirmed
from service_app_test.test_application.evyos.datas.get_occupants_codes import get_occupants_types
# from service_app_test.test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
decision_book_items_dict = (
@@ -17,7 +17,7 @@ decision_book_items_dict = (
)
def create_decision_book_items(decision_book_items):
def create_decision_book_items(decision_book_items, requester):
response = requester.post(
endpoint="/build/decision_book/items/create",
data=decision_book_items,
@@ -27,7 +27,7 @@ def create_decision_book_items(decision_book_items):
return response.json()
def send_invitation_to_building_residents(send_invitation_dict):
def send_invitation_to_building_residents(send_invitation_dict, requester):
response = requester.post(
endpoint="/build/decision_book/invite/create",
data=send_invitation_dict,
@@ -38,14 +38,14 @@ def send_invitation_to_building_residents(send_invitation_dict):
def assign_people_to_pre_or_wrt(
person_uu_id: str, manager_token: str, occupant_code: str
person_uu_id: str, manager_token: str, occupant_code: str, requester
):
response = requester.post(
endpoint="/build/decision_book/invitations/assign",
data={
"token": manager_token,
"build_living_space_uu_id": person_uu_id,
"occupant_type_uu_id": get_occupants_types(occupant_code=occupant_code)[
"occupant_type_uu_id": get_occupants_types(occupant_code=occupant_code, requester=requester)[
"data"
]["uu_id"],
},
@@ -56,7 +56,7 @@ def assign_people_to_pre_or_wrt(
def collect_invitation_to_building_residents(
attended_dict_list: list, attend_count: int
attended_dict_list: list, attend_count: int, requester
):
for cnt, attended_token in enumerate(attended_dict_list):
attended_dict = {"token": attended_token, "is_attend": False}
@@ -72,6 +72,7 @@ def collect_invitation_to_building_residents(
def create_decision_book_items_with_occupant_user(
requester,
writers_token: str,
unit_price: float,
is_fixed: bool,
@@ -93,12 +94,12 @@ def create_decision_book_items_with_occupant_user(
]
for item in list_of_items:
print("item", item)
create_decision_book_items(item)
create_decision_book_items(item, requester=requester)
return
def run_decision_book_items(
writers_token, unit_price, info_type_uu_id, is_fixed, start_date=None, end_date=None
writers_token, unit_price, info_type_uu_id, is_fixed, requester, start_date=None, end_date=None
):
if start_date and end_date:
create_decision_book_items_with_occupant_user(
@@ -108,6 +109,7 @@ def run_decision_book_items(
info_type_uu_id=info_type_uu_id,
start_date=start_date,
end_date=end_date,
requester=requester
)
else:
create_decision_book_items_with_occupant_user(
@@ -115,4 +117,5 @@ def run_decision_book_items(
unit_price=unit_price,
is_fixed=is_fixed,
info_type_uu_id=info_type_uu_id,
requester=requester
)

2
service_app_test/test_application/evyos/people.py
View File

@@ -1,7 +1,7 @@
import random
from datetime import datetime, timedelta
from validations import InsertPerson, InsertUsers
from api_validations.validations_request import InsertPerson, InsertUsers
from faker import Faker
faker_engine = Faker(["tr-TR"])

52
service_app_test/test_application/evyos/runner.py
View File

@@ -1,26 +1,29 @@
import arrow
from test_application.evyos.bases import login_via_email_and_password
from test_application.evyos.company_employee import run_company_and_depends
from test_application.evyos.address_building import run_address_to_building
from test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
from test_application.evyos.decision_book import (
from service_app_test.bases import RequestToApi
from service_app_test.api_configs import WagAPI, LocalAPI, BothAPIS
from service_app_test.test_application.evyos.company_employee import run_company_and_depends
from service_app_test.test_application.evyos.address_building import run_address_to_building
from service_app_test.test_application.evyos.datas.get_type_codes import get_type_codes_key_and_class
from service_app_test.test_application.evyos.decision_book import (
run_decision_book_depends,
list_decision_books,
)
from test_application.evyos.decision_book_items import (
from service_app_test.test_application.evyos.decision_book_items import (
run_decision_book_items,
send_invitation_to_building_residents,
collect_invitation_to_building_residents,
assign_people_to_pre_or_wrt,
)
login_creds_employee = {
"domain": "evyos.com.tr",
"access_key": "karatay.berkay.sup@evyos.com.tr",
"password": "string",
"remember_me": False,
"password_token": "",
"password_token": "z5nPguDu6PrY_94KlTGxGBCMcK6rZcDl5AGqO5aHWY7TcQF8LJGYikMQs9labvqdz5yM7vE_f8Zq_vqp00o3EnWwWCWhel-EsMcAujBKNRjV3jC_4xk0_1r_unIYogWAjz_spZipX6pncsch2ngvv0dpx4lHSiZ5QrDTA1RefQ"
}
access_key_president = "bmanco@example.net"
login_creds_occupant = {
@@ -38,14 +41,14 @@ wrt_creds_occupant = {
"password_token": "",
}
add_with_employee = None
add_with_occupant = True
assign_people_to_create_item = 3
add_with_employee = True
add_with_occupant = None
assign_people_to_create_item = 0
# selection_list = None
selection_list = [
"7e370616-7bcf-469f-b9a2-c0da55463939",
"c2d385d8-b772-4ecd-be89-8c468738654a",
# "7e370616-7bcf-469f-b9a2-c0da55463939",
# "c2d385d8-b772-4ecd-be89-8c468738654a",
]
# selection_list = None
manager_token = ""
@@ -53,16 +56,27 @@ writers_token = manager_token
people_uu_id_list = []
list_of_attendees = []
local_api = RequestToApi()
local_api.overwrite_base_url(base_url=LocalAPI.base_url)
if add_with_employee:
select_occupant = login_via_email_and_password(
print('local_api', local_api.post(endpoint="authentication/login", data=login_creds_employee).text)
exit()
local_api.selected_object = local_api.login_via_email_and_password(
login_data=login_creds_employee, is_password_valid=False
)
print("select_company_uu_id", select_occupant)
run_company_and_depends(company_uu_id=select_occupant)
run_address_to_building()
print('local_api', local_api.headers)
print("select_company_uu_id", local_api.selected_object)
exit()
run_company_and_depends(company_uu_id=local_api.selected_object, requester=local_api)
run_address_to_building(requester=local_api)
exit()
if add_with_occupant:
select_occupant = login_via_email_and_password(
local_api.selected_object = local_api.login_via_email_and_password(
login_data=login_creds_occupant,
is_password_valid=False,
selection_list=selection_list,
@@ -98,13 +112,13 @@ if add_with_occupant:
occupant_code="MT-PRS",
)
elif assign_people_to_create_item == 3:
select_occupant = login_via_email_and_password(
local_api.selected_object = local_api.login_via_email_and_password(
login_data=wrt_creds_occupant,
is_password_valid=True,
selection_list=selection_list,
)
selected_decision_book_uu_id = list_decision_books()["data"][0]["uu_id"]
print("select_company_uu_id", select_occupant)
print("select_company_uu_id", local_api.selected_object)
is_fixed_price = False
info_type_d_uu_id = get_type_codes_key_and_class(

0
service_app_test/test_application/migrate_old_data/accounts.py Normal file
View File

17
service_app_test/test_application/migrate_old_data/building.py Normal file
View File

@@ -0,0 +1,17 @@
from service_app_test.api_configs import BothAPIS
def migrate_build(requester: BothAPIS):
return
def migrate_build_area(requester: BothAPIS):
return
def migrate_build_part(requester: BothAPIS):
return
def migrate_build_iban(requester: BothAPIS):
return
def migrate_build_living_space(requester: BothAPIS):
return

31
service_app_test/test_application/migrate_old_data/company.py Normal file
View File

@@ -0,0 +1,31 @@
from service_app_test.api_configs import BothAPIS
from service_app_test.bases import FilterObject
def migrate_company(requester: BothAPIS):
filter_object = FilterObject(
page=1,
size=100,
order_field="id",
order_type="asc",
)
response = requester.wag_api.post(
endpoint="/company/list",
data=filter_object.dump(),
)
response_json = response.json()
print('response_json', response_json)
response_datas = response_json['data']
for response_data in response_datas:
print('response_data', response_data)
response_data['active'] = True
response_data['deleted'] = False
response_data['is_confirmed'] = True
response_data['expiry_starts'] = None
response_data['expiry_ends'] = None
response = requester.local_api.post(
endpoint="/company/create",
data=response_data,
)
print('response', response.text)
return

6
service_app_test/test_application/migrate_old_data/people.py Normal file
View File

@@ -0,0 +1,6 @@
from service_app_test.api_configs import BothAPIS
def migrate_people(requester: BothAPIS):
# Migrate old data
pass

39
service_app_test/test_application/migrate_old_data/reader_and_alchemy_bulk_actions.py Normal file
View File

@@ -0,0 +1,39 @@
from time import perf_counter
from json import loads
system_default = dict(
is_confirmed=True,
active=True,
deleted=False,
is_notification_send=True,
created_by="System",
confirmed_by="System",
)
def read_json_file(json_directory, json_file):
with open(f"{json_directory}/{json_file}.json", "r", encoding='utf-8') as json_file:
return loads(json_file.read())
def bulk_insert_alchemy_postgresql(active_session, table, data_list, on_conflict_constraints = None):
from sqlalchemy.dialects.postgresql import insert
st = perf_counter()
if on_conflict_constraints:
session_execute = active_session.execute(
insert(table)
.values(data_list)
.on_conflict_do_nothing(
index_elements=on_conflict_constraints or [],
)
)
else:
session_execute = active_session.execute(
insert(table)
.values(data_list)
)
count_row = session_execute.rowcount
print(f'Table : {table.__name__} count_row : {count_row} : time took : {round(perf_counter() - st, 2)} seconds')
active_session.commit()
active_session.flush()

52
service_app_test/test_application/migrate_old_data/runner.py Normal file
View File

@@ -0,0 +1,52 @@
from service_app_test.bases import RequestToApi
from service_app_test.api_configs import WagAPI, LocalAPI, BothAPIS
from service_app_test.test_application.migrate_old_data.people import migrate_people
from service_app_test.test_application.migrate_old_data.building import (
migrate_build,
migrate_build_area,
migrate_build_part,
migrate_build_iban,
migrate_build_living_space,
)
from service_app_test.test_application.migrate_old_data.company import migrate_company
login_data = {
"domain": "evyos.com.tr",
"access_key": "karatay.berkay.sup@evyos.com.tr",
"password": "string",
"remember_me": False,
"password_token": "z5nPguDu6PrY_94KlTGxGBCMcK6rZcDl5AGqO5aHWY7TcQF8LJGYikMQs9labvqdz5yM7vE_f8Zq_vqp00o3EnWwWCWhel-EsMcAujBKNRjV3jC_4xk0_1r_unIYogWAjz_spZipX6pncsch2ngvv0dpx4lHSiZ5QrDTA1RefQ",
}
login_data_wag = {
"domain": "evyos.com.tr",
"access_key": "karatay.berkay.sup@evyos.com.tr",
"password": "string",
"remember_me": False,
"password_token": ""
}
wag_api = RequestToApi()
wag_api.overwrite_base_url(base_url=WagAPI.base_url)
wag_api.selected_object = wag_api.login_via_email_and_password(
login_data=login_data_wag, is_password_valid=True
)
local_api = RequestToApi()
local_api.overwrite_base_url(base_url=LocalAPI.base_url)
local_api.selected_object = local_api.login_via_email_and_password(
login_data=login_data, is_password_valid=False
)
both_apis = BothAPIS()
both_apis.wag_api = wag_api
both_apis.local_api = local_api
migrate_company(requester=both_apis)
migrate_people(requester=both_apis)
migrate_build(requester=both_apis)
migrate_build_area(requester=both_apis)
migrate_build_part(requester=both_apis)
migrate_build_iban(requester=both_apis)
migrate_build_living_space(requester=both_apis)