urschrei.eurosky.social / pyzotero
Pyzotero: a Python client for the Zotero API pyzotero.readthedocs.io
zotero
fork atom

More upload and attachment tests

urschrei.eurosky.social 54f2acca a18874de

+298
+298
tests/test_zotero.py
··· 859 859 860 860 self.assertTrue(resp) 861 861 862 + @httpretty.activate 863 + def testFileUpload(self): 864 + """Tests file upload process with attachments""" 865 + zot = z.Zotero("myuserID", "user", "myuserkey") 866 + 867 + # Create a temporary file for testing 868 + temp_file_path = os.path.join(self.cwd, "api_responses", "test_upload_file.txt") 869 + with open(temp_file_path, "w") as f: 870 + f.write("Test file content for upload") 871 + 872 + # Mock Step 0: Create preliminary item registration 873 + prelim_response = { 874 + "success": { 875 + "0": "ITEMKEY123" 876 + } 877 + } 878 + HTTPretty.register_uri( 879 + HTTPretty.POST, 880 + "https://api.zotero.org/users/myuserID/items", 881 + content_type="application/json", 882 + body=json.dumps(prelim_response), 883 + status=200, 884 + ) 885 + 886 + # Create the upload payload 887 + payload = [{"filename": "test_upload_file.txt", "title": "Test File", "linkMode": "imported_file"}] 888 + 889 + # Create mock auth data to be returned by _get_auth 890 + mock_auth_data = { 891 + "url": "https://uploads.zotero.org/", 892 + "params": { 893 + "key": "abcdef1234567890", 894 + "prefix": "prefix", 895 + "suffix": "suffix" 896 + }, 897 + "uploadKey": "upload_key_123" 898 + } 899 + 900 + # Patch the necessary methods to avoid HTTP calls and file system checks 901 + with patch.object(z.Zupload, '_verify', return_value=None), \ 902 + patch.object(z.Zupload, '_get_auth', return_value=mock_auth_data), \ 903 + patch.object(z.Zupload, '_upload_file', return_value=None): 904 + # Create the upload object and initiate upload 905 + upload = z.Zupload(zot, payload, basedir=os.path.join(self.cwd, "api_responses")) 906 + result = upload.upload() 907 + 908 + # Verify the result structure 909 + self.assertIn("success", result) 910 + self.assertIn("failure", result) 911 + self.assertIn("unchanged", result) 912 + self.assertEqual(len(result["success"]), 1) 913 + self.assertEqual(result["success"][0]["key"], "ITEMKEY123") 914 + 915 + # Clean up 916 + os.remove(temp_file_path) 917 + 918 + @httpretty.activate 919 + def testFileUploadExists(self): 920 + """Tests file upload process when the file already exists on the server""" 921 + zot = z.Zotero("myuserID", "user", "myuserkey") 922 + 923 + # Create a temporary file for testing 924 + temp_file_path = os.path.join(self.cwd, "api_responses", "test_upload_file.txt") 925 + with open(temp_file_path, "w") as f: 926 + f.write("Test file content for upload") 927 + 928 + # Mock Step 0: Create preliminary item registration 929 + prelim_response = { 930 + "success": { 931 + "0": "ITEMKEY123" 932 + } 933 + } 934 + HTTPretty.register_uri( 935 + HTTPretty.POST, 936 + "https://api.zotero.org/users/myuserID/items", 937 + content_type="application/json", 938 + body=json.dumps(prelim_response), 939 + status=200, 940 + ) 941 + 942 + # Create the upload payload 943 + payload = [{"filename": "test_upload_file.txt", "title": "Test File", "linkMode": "imported_file"}] 944 + 945 + # Create mock auth data to be returned by _get_auth with exists=True 946 + mock_auth_data = { 947 + "exists": True 948 + } 949 + 950 + # Patch the necessary methods to avoid HTTP calls and file system checks 951 + with patch.object(z.Zupload, '_verify', return_value=None), \ 952 + patch.object(z.Zupload, '_get_auth', return_value=mock_auth_data): 953 + # Create the upload object and initiate upload 954 + upload = z.Zupload(zot, payload, basedir=os.path.join(self.cwd, "api_responses")) 955 + result = upload.upload() 956 + 957 + # Verify the result structure 958 + self.assertIn("success", result) 959 + self.assertIn("failure", result) 960 + self.assertIn("unchanged", result) 961 + self.assertEqual(len(result["unchanged"]), 1) 962 + self.assertEqual(result["unchanged"][0]["key"], "ITEMKEY123") 963 + 964 + # Clean up 965 + os.remove(temp_file_path) 966 + 967 + @httpretty.activate 968 + def testFileUploadWithParentItem(self): 969 + """Tests file upload process with a parent item ID""" 970 + zot = z.Zotero("myuserID", "user", "myuserkey") 971 + 972 + # Create a temporary file for testing 973 + temp_file_path = os.path.join(self.cwd, "api_responses", "test_upload_file.txt") 974 + with open(temp_file_path, "w") as f: 975 + f.write("Test file content for upload") 976 + 977 + # Mock Step 0: Create preliminary item registration 978 + prelim_response = { 979 + "success": { 980 + "0": "ITEMKEY123" 981 + } 982 + } 983 + HTTPretty.register_uri( 984 + HTTPretty.POST, 985 + "https://api.zotero.org/users/myuserID/items", 986 + content_type="application/json", 987 + body=json.dumps(prelim_response), 988 + status=200, 989 + ) 990 + 991 + # Create the upload payload 992 + payload = [{"filename": "test_upload_file.txt", "title": "Test File", "linkMode": "imported_file"}] 993 + 994 + # Test with parent ID 995 + parent_id = "PARENTITEM123" 996 + 997 + # Create mock auth data to be returned by _get_auth 998 + mock_auth_data = { 999 + "url": "https://uploads.zotero.org/", 1000 + "params": { 1001 + "key": "abcdef1234567890", 1002 + "prefix": "prefix", 1003 + "suffix": "suffix" 1004 + }, 1005 + "uploadKey": "upload_key_123" 1006 + } 1007 + 1008 + # Mock Step 1: Get upload authorization 1009 + HTTPretty.register_uri( 1010 + HTTPretty.POST, 1011 + "https://api.zotero.org/users/myuserID/items/ITEMKEY123/file", 1012 + content_type="application/json", 1013 + body=json.dumps(mock_auth_data), 1014 + status=200, 1015 + ) 1016 + 1017 + # Patch the necessary methods to avoid file system checks and skip the actual upload 1018 + with patch.object(z.Zupload, '_verify', return_value=None), \ 1019 + patch.object(z.Zupload, '_upload_file', return_value=None): 1020 + # Create the upload object with a parent ID and initiate upload 1021 + upload = z.Zupload(zot, payload, parentid=parent_id, basedir=os.path.join(self.cwd, "api_responses")) 1022 + result = upload.upload() 1023 + 1024 + # Verify the result structure 1025 + self.assertIn("success", result) 1026 + self.assertEqual(len(result["success"]), 1) 1027 + 1028 + # Check that the parentItem was added to the payload 1029 + # Get the latest request to the items endpoint 1030 + requests = httpretty.latest_requests() 1031 + item_request = None 1032 + for req in requests: 1033 + if req.url.endswith("/items"): 1034 + item_request = req 1035 + break 1036 + 1037 + self.assertIsNotNone(item_request, "No request found to the items endpoint") 1038 + request_body = json.loads(item_request.body.decode('utf-8')) 1039 + self.assertEqual(request_body[0]["parentItem"], parent_id) 1040 + 1041 + # Clean up 1042 + os.remove(temp_file_path) 1043 + 1044 + @httpretty.activate 1045 + def testFileUploadFailure(self): 1046 + """Tests file upload process when auth step fails""" 1047 + zot = z.Zotero("myuserID", "user", "myuserkey") 1048 + 1049 + # Create a temporary file for testing 1050 + temp_file_path = os.path.join(self.cwd, "api_responses", "test_upload_file.txt") 1051 + with open(temp_file_path, "w") as f: 1052 + f.write("Test file content for upload") 1053 + 1054 + # Mock Step 0: Create preliminary item registration 1055 + prelim_response = { 1056 + "success": { 1057 + "0": "ITEMKEY123" 1058 + } 1059 + } 1060 + HTTPretty.register_uri( 1061 + HTTPretty.POST, 1062 + "https://api.zotero.org/users/myuserID/items", 1063 + content_type="application/json", 1064 + body=json.dumps(prelim_response), 1065 + status=200, 1066 + ) 1067 + 1068 + # Mock Step 1: Authorization fails with 403 1069 + HTTPretty.register_uri( 1070 + HTTPretty.POST, 1071 + "https://api.zotero.org/users/myuserID/items/ITEMKEY123/file", 1072 + status=403, 1073 + ) 1074 + 1075 + # Create the upload payload 1076 + payload = [{"filename": "test_upload_file.txt", "title": "Test File", "linkMode": "imported_file"}] 1077 + 1078 + # Patch just the _verify method to avoid file system checks, but allow the real HTTP calls 1079 + with patch.object(z.Zupload, '_verify', return_value=None): 1080 + # Create the upload object and test for exception 1081 + upload = z.Zupload(zot, payload, basedir=os.path.join(self.cwd, "api_responses")) 1082 + 1083 + # This should raise an error due to 403 status 1084 + with self.assertRaises(z.ze.UserNotAuthorisedError): 1085 + upload.upload() 1086 + 1087 + # Clean up 1088 + os.remove(temp_file_path) 1089 + 1090 + @httpretty.activate 1091 + def testFileUploadWithPreexistingKeys(self): 1092 + """Tests file upload process when the payload already contains keys""" 1093 + zot = z.Zotero("myuserID", "user", "myuserkey") 1094 + 1095 + # Create a temporary file for testing 1096 + temp_file_path = os.path.join(self.cwd, "api_responses", "test_upload_file.txt") 1097 + with open(temp_file_path, "w") as f: 1098 + f.write("Test file content for upload") 1099 + 1100 + # Create the upload payload with preexisting key 1101 + payload = [{"key": "PREEXISTING123", "filename": "test_upload_file.txt", "title": "Test File", "linkMode": "imported_file"}] 1102 + 1103 + # Create mock auth data to be returned by _get_auth 1104 + mock_auth_data = { 1105 + "url": "https://uploads.zotero.org/", 1106 + "params": { 1107 + "key": "abcdef1234567890", 1108 + "prefix": "prefix", 1109 + "suffix": "suffix" 1110 + }, 1111 + "uploadKey": "upload_key_123" 1112 + } 1113 + 1114 + # Patch the necessary methods to avoid HTTP calls and file system checks 1115 + with patch.object(z.Zupload, '_verify', return_value=None), \ 1116 + patch.object(z.Zupload, '_get_auth', return_value=mock_auth_data), \ 1117 + patch.object(z.Zupload, '_upload_file', return_value=None): 1118 + # Create the upload object and initiate upload 1119 + upload = z.Zupload(zot, payload, basedir=os.path.join(self.cwd, "api_responses")) 1120 + result = upload.upload() 1121 + 1122 + # Verify the result structure 1123 + self.assertIn("success", result) 1124 + self.assertEqual(len(result["success"]), 1) 1125 + self.assertEqual(result["success"][0]["key"], "PREEXISTING123") 1126 + 1127 + # No need to check for endpoint calls since we're patching the methods 1128 + 1129 + # Clean up 1130 + os.remove(temp_file_path) 1131 + 1132 + @httpretty.activate 1133 + def testFileUploadInvalidPayload(self): 1134 + """Tests file upload process with invalid payload mixing items with and without keys""" 1135 + zot = z.Zotero("myuserID", "user", "myuserkey") 1136 + 1137 + # Create a temporary file for testing 1138 + temp_file_path = os.path.join(self.cwd, "api_responses", "test_upload_file.txt") 1139 + with open(temp_file_path, "w") as f: 1140 + f.write("Test file content for upload") 1141 + 1142 + # Create the invalid upload payload (mixing items with and without keys) 1143 + payload = [ 1144 + {"key": "PREEXISTING123", "filename": "test_upload_file.txt"}, 1145 + {"filename": "test_upload_file.txt"} # No key 1146 + ] 1147 + 1148 + # Patch the _verify method to avoid file system checks 1149 + with patch.object(z.Zupload, '_verify', return_value=None): 1150 + # Create the upload object and test for exception 1151 + upload = z.Zupload(zot, payload, basedir=os.path.join(self.cwd, "api_responses")) 1152 + 1153 + # This should raise an UnsupportedParamsError 1154 + with self.assertRaises(z.ze.UnsupportedParamsError): 1155 + upload.upload() 1156 + 1157 + # Clean up 1158 + os.remove(temp_file_path) 1159 + 862 1160 def tearDown(self): 863 1161 """Tear stuff down""" 864 1162 HTTPretty.disable()