pranavshuklaa commented on code in PR #1091:
URL: https://github.com/apache/flink-agents/pull/1091#discussion_r4543832110
##########
python/flink_agents/runtime/skill/tests/test_url_repository.py:
##########
@@ -208,3 +212,34 @@ def test_404_error(self, zip_server: "tuple[str,
type[_ZipHandler]]") -> None:
with pytest.raises(HTTPError) as exc_info:
URLSkillRepository(f"{url}?token=top-secret",
allow_insecure_http=True)
assert "top-secret" not in str(exc_info.value)
+
+
+class TestURLSkillRepositoryLimits:
+ def test_download_uses_given_limits(
+ self, zip_server: "tuple[str, type[_ZipHandler]]"
+ ) -> None:
+ url, handler = zip_server
+ size = len(handler.zip_bytes)
+ limits = MaterializerLimits(
+ max_download_bytes=size - 1,
+ max_extract_entry_bytes=DEFAULT_LIMITS.max_extract_entry_bytes,
+ max_extract_total_bytes=DEFAULT_LIMITS.max_extract_total_bytes,
+ max_extract_entries=DEFAULT_LIMITS.max_extract_entries,
+ )
+ with pytest.raises(ValueError, match=f"limit of {size - 1} bytes"):
+ URLSkillRepository(url, allow_insecure_http=True, limits=limits)
+ # Control: the same archive loads with the default limits.
+ assert URLSkillRepository(url, allow_insecure_http=True).get_skills()
Review Comment:
Done. Both successful control cases, URL and filesystem zip, now close their
repository in a `finally` block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
