uranusjr commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4491148441
##########
providers/amazon/tests/unit/amazon/aws/bundles/test_s3.py:
##########
@@ -262,3 +282,265 @@ def test_refresh_uses_directory_prefix(self, s3_client,
prefix, extra_suffix):
assert repr(bundle) == original_repr
assert bundle.view_url_template() == original_url
assert s3_client.get_object(Bucket=S3_BUCKET_NAME,
Key=extra_key)["Body"].read() == b"outside"
+
+
[email protected](not airflow.version.version.strip().startswith("3"),
reason="Airflow >=3.0.0 test")
+class TestS3ArchiveDagBundle:
+ def _archive_bundle(self) -> S3ArchiveDagBundle:
+ return S3ArchiveDagBundle(
+ name="test",
+ aws_conn_id=AWS_CONN_ID_WITH_REGION,
+ archive_key=S3_ARCHIVE_KEY,
+ bucket_name=S3_BUCKET_NAME,
+ )
+
+ def test_view_url_template_points_at_archive(self):
+ bundle = self._archive_bundle()
+ assert (
+ bundle.view_url_template()
+ ==
f"https://{S3_BUCKET_NAME}.s3.{AWS_CONN_ID_REGION}.amazonaws.com/{S3_ARCHIVE_KEY}";
+ )
+
+ def test_supports_versioning(self):
+ bundle = self._archive_bundle()
+ assert S3ArchiveDagBundle.supports_versioning is False
+ bundle.version = "test_version"
+ with pytest.raises(AirflowException, match="Refreshing a specific
version is not supported"):
+ bundle.refresh()
+ with pytest.raises(AirflowException, match="S3 url with version is not
supported"):
+ bundle.view_url("test_version")
+
+ def test_refresh_from_archive(self, s3_bucket, s3_client):
+ archive = _make_archive({"dag_01.py": b"test data",
"subproject1/dag_a.py": b"test data"})
+ s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=S3_ARCHIVE_KEY,
Body=archive)
+
+ bundle = self._archive_bundle()
+ bundle.initialize()
+
+ assert (bundle.path / "dag_01.py").is_file()
+ assert (bundle.path / "subproject1" / "dag_a.py").is_file()
+ assert bundle._get_archive_etag_path().is_file()
+ assert sorted(path.relative_to(bundle.path).as_posix() for path in
bundle.path.rglob("*")) == [
+ "dag_01.py",
+ "subproject1",
+ "subproject1/dag_a.py",
+ ]
+ assert not (bundle.path / S3_BUCKET_PREFIX / "dag_02.py").exists()
+ assert not (bundle.path / "dag_02.py").exists()
+
+ def test_refresh_from_archive_skips_staging_when_etag_unchanged(self,
s3_bucket, s3_client):
+ archive = _make_archive({"dag_01.py": b"test data"})
+ s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=S3_ARCHIVE_KEY,
Body=archive)
+
+ bundle = self._archive_bundle()
+ bundle.initialize()
+
+ sentinel = bundle.path / "sentinel.txt"
+ sentinel.write_text("still here")
+ bundle.refresh()
+ assert sentinel.is_file()
+
+ updated = _make_archive({"dag_new.py": b"test data"})
+ s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=S3_ARCHIVE_KEY,
Body=updated)
+ bundle.refresh()
+ assert (bundle.path / "dag_new.py").is_file()
+ assert not (bundle.path / "dag_01.py").exists()
+ assert not sentinel.exists()
+
+ def
test_initialize_restages_when_bundle_dir_cleared_and_etag_matches(self,
s3_bucket, s3_client):
+ archive = _make_archive({"dag_01.py": b"test data"})
+ s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=S3_ARCHIVE_KEY,
Body=archive)
+
+ bundle = self._archive_bundle()
+ bundle.initialize()
+ marker = bundle._get_archive_etag_path()
+ etag = marker.read_text()
+ assert (bundle.path / "dag_01.py").is_file()
+
+ shutil.rmtree(bundle.path)
+ assert not bundle.path.exists()
Review Comment:
This `assert` simply checked `shutil.rmtree` works. That’s not our code.
--
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]
