fix azure tests

64;2500;0c
> Insert changelog entry or delete this section.

Don't delete keys before overwriting in azure blob IO manager

Summary:
The fact that this sets overwrite=True makes me strongly suspect that this is safe, but the relevant test is disabled :( https://github.com/dagster-io/dagster/blob/97b7bfae8d9dbfb49daaf76c58932dd2fb55b6bd/python_modules/libraries/dagster-azure/dagster_azure_tests/adls2_tests/test_io_manager.py#L237-L242 Need to figure out how to get it working.

TBD

> Insert changelog entry or delete this section.

Author: Daniel Gibson

python_modules/libraries/dagster-azure/dagster_azure/adls2/io_manager.py

@@ -105,10 +105,6 @@ def load_from_path(self, context: InputContext, path: UPath) -> Any:
         return pickle.loads(stream.readall())
 
     def dump_to_path(self, context: OutputContext, obj: Any, path: UPath) -> None:
-        if self.path_exists(path):
-            context.log.warning(f"Removing existing ADLS2 key: {path}")
-            self.unlink(path)
-
         pickled_obj = pickle.dumps(obj, PICKLE_PROTOCOL)
         file = self.file_system_client.create_file(path.as_posix())
         with self._acquire_lease(file) as lease:

python_modules/libraries/dagster-azure/dagster_azure_tests/adls2_tests/test_io_manager.py

@@ -23,8 +23,6 @@
 from dagster._core.definitions.assets.definition.assets_definition import AssetsDefinition
 from dagster._core.definitions.definitions_class import Definitions
 from dagster._core.definitions.job_base import InMemoryJob
-from dagster._core.definitions.partitions.definition import StaticPartitionsDefinition
-from dagster._core.definitions.source_asset import SourceAsset
 from dagster._core.definitions.unresolved_asset_job_definition import define_asset_job
 from dagster._core.events import DagsterEventType
 from dagster._core.execution.api import create_execution_plan, execute_plan
@@ -81,10 +79,6 @@ def basic_external_plan_execution():
 
 
 @pytest.mark.nettest
-@pytest.mark.skip(
-    "Blob this depends on does not exist. See"
-    " https://linear.app/elementl/issue/CORE-83/test-adls2-pickle-io-manager-deletes-recursively-disabled-reenable-it"
-)
 def test_adls2_pickle_io_manager_deletes_recursively(storage_account, file_system, credential):
     job = define_inty_job()
 
@@ -151,10 +145,6 @@ def test_adls2_pickle_io_manager_deletes_recursively(storage_account, file_syste
 
 
 @pytest.mark.nettest
-@pytest.mark.skip(
-    "Blob this depends on does not exist. See"
-    " https://linear.app/elementl/issue/CORE-83/test-adls2-pickle-io-manager-deletes-recursively-disabled-reenable-it"
-)
 def test_adls2_pickle_io_manager_execution(storage_account, file_system, credential):
     job = define_inty_job()
 
@@ -235,10 +225,7 @@ def test_adls2_pickle_io_manager_execution(storage_account, file_system, credent
     assert io_manager.load_input(context) == 2
 
 
-@pytest.mark.skip(
-    "Blob this depends on does not exist. See"
-    " https://linear.app/elementl/issue/CORE-83/test-adls2-pickle-io-manager-deletes-recursively-disabled-reenable-it"
-)
+@pytest.mark.nettest
 def test_asset_io_manager(storage_account, file_system, credential):
     # if you add new assets to this test, make sure that the output names include _id so that we don't
     # run into issues with the azure leasing system in CI
@@ -261,37 +248,20 @@ def graph_asset():
 
     @asset(
         name=f"upstream_{_id}",
-        ins={"asset3": AssetIn(asset_key=AssetKey([f"asset3_{_id}"]))},  # pyright: ignore[reportCallIssue]
+        ins={"asset3": AssetIn(key=AssetKey([f"asset3_{_id}"]))},
     )
     def upstream(asset3):
         return asset3 + 1
 
-    SourceAsset(f"source1_{_id}", partitions_def=StaticPartitionsDefinition(["foo", "bar"]))
-
-    # prepopulate storage with source asset
-    io_manager = PickledObjectADLS2IOManager(
-        file_system=file_system,
-        adls2_client=create_adls2_client(storage_account, credential),
-        blob_client=create_blob_client(storage_account, credential),
-        lease_client_constructor=DataLakeLeaseClient,
-    )
-    for partition_key in ["foo", "bar"]:
-        context = build_output_context(
-            step_key=f"source1_{_id}",
-            name="result",
-            run_id=make_new_run_id(),
-            dagster_type=resolve_dagster_type(int),
-            partition_key=partition_key,
-        )
-        io_manager.handle_output(context, 1)
-
     @asset(
         name=f"downstream_{_id}",
-        ins={"upstream": AssetIn(asset_key=AssetKey([f"upstream_{_id}"]))},  # pyright: ignore[reportCallIssue]
+        ins={
+            "upstream": AssetIn(key=AssetKey([f"upstream_{_id}"])),
+        },
     )
-    def downstream(upstream, source):
+    def downstream(upstream):
         assert upstream == 7
-        return 1 + upstream + source["foo"] + source["bar"]
+        return 1 + upstream
 
     asset_job = Definitions(
         assets=[upstream, downstream, AssetsDefinition.from_graph(graph_asset)],
@@ -314,6 +284,11 @@ def downstream(upstream, source):
     result = asset_job.execute_in_process(run_config=run_config)
     assert result.success
 
+    # overwriting the same bucket still works
+
+    result = asset_job.execute_in_process(run_config=run_config)
+    assert result.success
+
 
 def test_with_fake_adls2_resource():
     job = define_inty_job(adls_io_resource=fake_adls2_resource)