This function is used to fetch all the schemas of source and target database and compare them. :param source_sid: :param source_did: :param target_sid: :param target_did: :return:
(source_sid, source_did, target_sid, target_did)
| 934 | |
| 935 | |
| 936 | def fetch_compare_schemas(source_sid, source_did, target_sid, target_did): |
| 937 | """ |
| 938 | This function is used to fetch all the schemas of source and target |
| 939 | database and compare them. |
| 940 | |
| 941 | :param source_sid: |
| 942 | :param source_did: |
| 943 | :param target_sid: |
| 944 | :param target_did: |
| 945 | :return: |
| 946 | """ |
| 947 | source_schemas = get_schemas(source_sid, source_did) |
| 948 | target_schemas = get_schemas(target_sid, target_did) |
| 949 | |
| 950 | if source_schemas is None or target_schemas is None: |
| 951 | return None |
| 952 | |
| 953 | src_schema_dict = {item['label']: item['_id'] for item in source_schemas} |
| 954 | tar_schema_dict = {item['label']: item['_id'] for item in target_schemas} |
| 955 | |
| 956 | dict1 = copy.deepcopy(src_schema_dict) |
| 957 | dict2 = copy.deepcopy(tar_schema_dict) |
| 958 | |
| 959 | # Find the duplicate keys in both the dictionaries |
| 960 | dict1_keys = set(dict1.keys()) |
| 961 | dict2_keys = set(dict2.keys()) |
| 962 | intersect_keys = dict1_keys.intersection(dict2_keys) |
| 963 | |
| 964 | # Keys that are available in source and missing in target. |
| 965 | source_only = [] |
| 966 | added = dict1_keys - dict2_keys |
| 967 | for item in added: |
| 968 | source_only.append({'schema_name': item, |
| 969 | 'scid': src_schema_dict[item]}) |
| 970 | |
| 971 | target_only = [] |
| 972 | # Keys that are available in target and missing in source. |
| 973 | removed = dict2_keys - dict1_keys |
| 974 | for item in removed: |
| 975 | target_only.append({'schema_name': item, |
| 976 | 'scid': tar_schema_dict[item]}) |
| 977 | |
| 978 | in_both_database = [] |
| 979 | for item in intersect_keys: |
| 980 | in_both_database.append({'schema_name': item, |
| 981 | 'src_scid': src_schema_dict[item], |
| 982 | 'tar_scid': tar_schema_dict[item]}) |
| 983 | |
| 984 | schema_result = {'source_only': source_only, 'target_only': target_only, |
| 985 | 'in_both_database': in_both_database} |
| 986 | |
| 987 | return schema_result |
| 988 | |
| 989 | |
| 990 | def compare_pre_validation(trans_id, source_sid, target_sid): |
no test coverage detected