Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 44 additions & 22 deletions sdk/cosmos/azure-cosmos/tests/test_full_text_policy.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# The MIT License (MIT)
# Copyright (c) Microsoft Corporation. All rights reserved.

import re
import unittest
import uuid

Expand Down Expand Up @@ -83,7 +84,8 @@ def test_create_full_text_container(self):
indexing_policy=indexing_policy
)
properties = created_container.read()
assert properties["fullTextPolicy"] == full_text_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == full_text_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == full_text_policy["fullTextPaths"]
assert properties["indexingPolicy"]['fullTextIndexes'] == indexing_policy['fullTextIndexes']
self.test_db.delete_container(created_container.id)

Expand All @@ -97,7 +99,7 @@ def test_create_full_text_container(self):
full_text_policy=full_text_policy_no_paths,
)
properties = created_container.read()
assert properties["fullTextPolicy"] == full_text_policy_no_paths
assert properties["fullTextPolicy"]["defaultLanguage"] == full_text_policy_no_paths["defaultLanguage"]
self.test_db.delete_container(created_container.id)

# Create a container with a full text policy with a given path containing only default language
Expand All @@ -115,7 +117,8 @@ def test_create_full_text_container(self):
full_text_policy=full_text_policy_no_langs,
)
properties = created_container.read()
assert properties["fullTextPolicy"] == full_text_policy_no_langs
assert properties["fullTextPolicy"]["defaultLanguage"] == full_text_policy_no_langs["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == full_text_policy_no_langs["fullTextPaths"]
self.test_db.delete_container(created_container.id)

def test_replace_full_text_container(self):
Expand Down Expand Up @@ -149,7 +152,8 @@ def test_replace_full_text_container(self):
indexing_policy=indexing_policy
)
properties = replaced_container.read()
assert properties["fullTextPolicy"] == full_text_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == full_text_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == full_text_policy["fullTextPaths"]
assert properties["indexingPolicy"]['fullTextIndexes'] == indexing_policy['fullTextIndexes']
assert created_container_properties['indexingPolicy'] != properties['indexingPolicy']
self.test_db.delete_container(created_container.id)
Expand All @@ -162,8 +166,9 @@ def test_replace_full_text_container(self):
indexing_policy=indexing_policy
)
created_container_properties = created_container.read()
assert properties["fullTextPolicy"] == full_text_policy
assert properties["indexingPolicy"]['fullTextIndexes'] == indexing_policy['fullTextIndexes']
assert created_container_properties["fullTextPolicy"]["defaultLanguage"] == full_text_policy["defaultLanguage"]
assert created_container_properties["fullTextPolicy"]["fullTextPaths"] == full_text_policy["fullTextPaths"]
assert created_container_properties["indexingPolicy"]['fullTextIndexes'] == indexing_policy['fullTextIndexes']

# Replace the container with new policies
full_text_policy['fullTextPaths'][0]['path'] = "/new_path"
Expand All @@ -175,7 +180,8 @@ def test_replace_full_text_container(self):
indexing_policy=indexing_policy
)
properties = replaced_container.read()
assert properties["fullTextPolicy"] == full_text_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == full_text_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == full_text_policy["fullTextPaths"]
assert properties["indexingPolicy"]['fullTextIndexes'] == indexing_policy['fullTextIndexes']
assert created_container_properties['fullTextPolicy'] != properties['fullTextPolicy']
assert created_container_properties["indexingPolicy"] != properties["indexingPolicy"]
Expand All @@ -201,7 +207,9 @@ def test_fail_create_full_text_policy(self):
pytest.fail("Container creation should have failed for invalid path.")
except exceptions.CosmosHttpResponseError as e:
assert e.status_code == 400
assert "The Full Text Policy contains an invalid Path: abstract" in e.http_error_message
assert re.search(
r"the full.text policy contains an invalid path.*abstract",
e.http_error_message, re.IGNORECASE)

# Pass a full text policy with an unsupported default language
full_text_policy_wrong_default = {
Expand All @@ -222,8 +230,9 @@ def test_fail_create_full_text_policy(self):
pytest.fail("Container creation should have failed for wrong supported language.")
except exceptions.CosmosHttpResponseError as e:
assert e.status_code == 400
assert "The Full Text Policy contains an unsupported language spa-SPA. Supported languages are:"\
in e.http_error_message
assert re.search(
r"the full.text policy contains an unsupported language.*spa-SPA",
e.http_error_message, re.IGNORECASE)

# Pass a full text policy with an unsupported path language
full_text_policy_wrong_default = {
Expand All @@ -244,8 +253,9 @@ def test_fail_create_full_text_policy(self):
pytest.fail("Container creation should have failed for wrong supported language.")
except exceptions.CosmosHttpResponseError as e:
assert e.status_code == 400
assert "The Full Text Policy contains an unsupported language spa-SPA. Supported languages are:"\
in e.http_error_message
assert re.search(
r"the full.text policy contains an unsupported language.*spa-SPA",
e.http_error_message, re.IGNORECASE)

def test_fail_create_full_text_indexing_policy(self):
full_text_policy = {
Expand Down Expand Up @@ -274,8 +284,9 @@ def test_fail_create_full_text_indexing_policy(self):
# pytest.fail("Container creation should have failed for lack of embedding policy.")
except exceptions.CosmosHttpResponseError as e:
assert e.status_code == 400
assert "The path of the Full Text Index /path does not match the path specified in the Full Text Policy"\
in e.http_error_message
assert re.search(
r"the path of the full.text index.*does not match the path specified in the full.text policy",
e.http_error_message, re.IGNORECASE)

# Pass a full text indexing policy with a wrongly formatted path
indexing_policy_wrong_path = {
Expand All @@ -293,7 +304,9 @@ def test_fail_create_full_text_indexing_policy(self):
pytest.fail("Container creation should have failed for invalid path.")
except exceptions.CosmosHttpResponseError as e:
assert e.status_code == 400
assert "Full-text index specification at index (0) contains invalid path" in e.http_error_message
assert re.search(
r"full.text index specification at index \(0\) contains invalid path",
e.http_error_message, re.IGNORECASE)

# Pass a full text indexing policy without a path field
indexing_policy_no_path = {
Expand All @@ -311,7 +324,9 @@ def test_fail_create_full_text_indexing_policy(self):
pytest.fail("Container creation should have failed for missing path.")
except exceptions.CosmosHttpResponseError as e:
assert e.status_code == 400
assert "Missing path in full-text index specification at index (0)" in e.http_error_message
assert re.search(
r"missing path in full.text index specification at index \(0\)",
e.http_error_message, re.IGNORECASE)

# Skipped until testing pipeline is set up for full text multi-language support
@pytest.mark.skip
Expand Down Expand Up @@ -342,7 +357,8 @@ def test_supported_languages_in_full_text_policy(self):
full_text_policy=updated_policy
)
properties = replaced_container.read()
assert properties["fullTextPolicy"] == updated_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == updated_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == updated_policy["fullTextPaths"]
finally:
self.test_db.delete_container(container.id)

Expand Down Expand Up @@ -376,7 +392,8 @@ def test_default_language_fallback(self):
full_text_policy=updated_policy
)
properties = replaced_container.read()
assert properties["fullTextPolicy"] == updated_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == updated_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == updated_policy["fullTextPaths"]
# Insert and verify item
item = {
"id": str(uuid.uuid4()),
Expand Down Expand Up @@ -429,7 +446,8 @@ def test_mismatched_default_and_path_languages(self):
full_text_policy=updated_policy
)
properties = replaced_container.read()
assert properties["fullTextPolicy"] == updated_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == updated_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == updated_policy["fullTextPaths"]

finally:
# Delete the container at the end of the test
Expand Down Expand Up @@ -472,7 +490,8 @@ def test_replace_full_text_policy_with_different_languages(self):
full_text_policy=updated_policy
)
properties = replaced_container.read()
assert properties["fullTextPolicy"] == updated_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == updated_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == updated_policy["fullTextPaths"]

finally:
# Delete the container at the end of the test
Expand Down Expand Up @@ -515,7 +534,8 @@ def test_replace_full_text_policy_with_different_path_languages(self):
full_text_policy=updated_policy
)
properties = replaced_container.read()
assert properties["fullTextPolicy"] == updated_policy
assert properties["fullTextPolicy"]["defaultLanguage"] == updated_policy["defaultLanguage"]
assert properties["fullTextPolicy"]["fullTextPaths"] == updated_policy["fullTextPaths"]

finally:
# Delete the container at the end of the test
Expand Down Expand Up @@ -611,7 +631,9 @@ def test_unsupported_language_in_full_text_policy(self):
pytest.fail("Container replacement should have failed for unsupported language.")
except exceptions.CosmosHttpResponseError as e:
assert e.status_code == 400
assert "The Full Text Policy contains an unsupported language" in e.http_error_message
assert re.search(
r"the full.text policy contains an unsupported language",
e.http_error_message, re.IGNORECASE)
finally:
self.test_db.delete_container(container.id)

Expand Down
Loading
Loading