Apply Eric's suggested changes.

# Rust tests and lints will be skipped. Delete if not intended.
[ci skip-rust]

# Building wheels and fs_util will be skipped. Delete if not intended.
[ci skip-build-wheels]

src/python/pants/engine/internals/graph_test.py

@@ -764,6 +764,10 @@ async def generate_smalltalk_from_avro(
 ) -> GeneratedSources:
     protocol_files = request.protocol_sources.files
 
+    # Many codegen implementations will need to look up a protocol target's dependencies in their
+    # rule. We add this here to ensure that this does not result in rule graph issues.
+    _ = await Get(TransitiveTargets, Addresses([request.protocol_target.address]))
+
     def generate_fortran(fp: str) -> FileContent:
         parent = str(PurePath(fp).parent).replace("src/avro", "src/smalltalk")
         file_name = f"{PurePath(fp).stem}.st"
@@ -800,25 +804,32 @@ def test_generate_sources(self) -> None:
 
         # First, get the original protocol sources.
         hydrated_protocol_sources = self.request_single_product(
-            HydratedSources, HydrateSourcesRequest(protocol_sources)
+            HydratedSources,
+            Params(HydrateSourcesRequest(protocol_sources), create_options_bootstrapper()),
         )
         assert hydrated_protocol_sources.snapshot.files == ("src/avro/f.avro",)
 
         # Test directly feeding the protocol sources into the codegen rule.
         wrapped_tgt = self.request_single_product(WrappedTarget, self.address)
         generated_sources = self.request_single_product(
             GeneratedSources,
-            GenerateSmalltalkFromAvroRequest(
-                hydrated_protocol_sources.snapshot, wrapped_tgt.target
+            Params(
+                GenerateSmalltalkFromAvroRequest(
+                    hydrated_protocol_sources.snapshot, wrapped_tgt.target
+                ),
+                create_options_bootstrapper(),
             ),
         )
         assert generated_sources.snapshot.files == ("src/smalltalk/f.st",)
 
         # Test that HydrateSourcesRequest can also be used.
         generated_via_hydrate_sources = self.request_single_product(
             HydratedSources,
-            HydrateSourcesRequest(
-                protocol_sources, for_sources_types=[SmalltalkSources], enable_codegen=True
+            Params(
+                HydrateSourcesRequest(
+                    protocol_sources, for_sources_types=[SmalltalkSources], enable_codegen=True
+                ),
+                create_options_bootstrapper(),
             ),
         )
         assert generated_via_hydrate_sources.snapshot.files == ("src/smalltalk/f.st",)
@@ -832,8 +843,11 @@ class CustomAvroSources(AvroSources):
         assert protocol_sources.can_generate(SmalltalkSources, self.union_membership) is True
         generated = self.request_single_product(
             HydratedSources,
-            HydrateSourcesRequest(
-                protocol_sources, for_sources_types=[SmalltalkSources], enable_codegen=True
+            Params(
+                HydrateSourcesRequest(
+                    protocol_sources, for_sources_types=[SmalltalkSources], enable_codegen=True
+                ),
+                create_options_bootstrapper(),
             ),
         )
         assert generated.snapshot.files == ("src/smalltalk/f.st",)
@@ -846,8 +860,11 @@ class AdaSources(Sources):
         assert protocol_sources.can_generate(AdaSources, self.union_membership) is False
         generated = self.request_single_product(
             HydratedSources,
-            HydrateSourcesRequest(
-                protocol_sources, for_sources_types=[AdaSources], enable_codegen=True
+            Params(
+                HydrateSourcesRequest(
+                    protocol_sources, for_sources_types=[AdaSources], enable_codegen=True
+                ),
+                create_options_bootstrapper(),
             ),
         )
         assert generated.snapshot.files == ()

src/python/pants/engine/target.py

@@ -648,23 +648,25 @@ def generate_subtarget(
     are able to deduce specifically which files are being used, we can use only the files we care
     about, rather than the entire `sources` field.
     """
-    if not base_target.has_field(Dependencies):
+    if not base_target.has_field(Dependencies) or not base_target.has_field(Sources):
         raise ValueError(
             f"Target {base_target.address.spec} of type {type(base_target).__qualname__} does "
-            f"not support dependencies, and thus cannot depend on {full_file_name}."
+            "not have both a `dependencies` and `sources` field, and thus cannot generate a "
+            f"subtarget for the file {full_file_name}."
         )
 
     relativized_file_name = (
         PurePath(full_file_name).relative_to(base_target.address.spec_path).as_posix()
     )
 
-    found_source_match = False
     generated_target_fields = {}
     for field in base_target.field_values.values():
-        if isinstance(field, Sources) and bool(
-            matches_filespec(field.filespec, paths=(full_file_name,))
-        ):
-            found_source_match = True
+        if isinstance(field, Sources):
+            if not bool(matches_filespec(field.filespec, paths=[full_file_name])):
+                raise ValueError(
+                    f"Target {base_target.address.spec}'s `sources` field does not match a file "
+                    f"{full_file_name}."
+                )
             value = (relativized_file_name,)
         else:
             value = (
@@ -674,18 +676,6 @@ def generate_subtarget(
             )
         generated_target_fields[field.alias] = value
 
-    if not found_source_match:
-        # TODO: This should probably be ResolveError, but this function is in the wrong place
-        # for that.
-        sources_fields = [
-            field for field in base_target.field_values.values() if isinstance(field, Sources)
-        ]
-        sources_fields_hint = f" It only matches: {sources_fields}." if sources_fields else ""
-        raise ValueError(
-            f"Target {base_target.address.spec} does not own "
-            f"a file {full_file_name}.{sources_fields_hint}"
-        )
-
     target_cls = type(base_target)
     return target_cls(
         generated_target_fields,

src/python/pants/engine/target_test.py

@@ -451,14 +451,21 @@ class MockTarget(Target):
         == expected_subdir_address
     )
 
-    class NoSourcesTgt(Target):
-        alias = "no_sources_tgt"
+    # The full_file_name must match the filespec of the base target's Sources field.
+    with pytest.raises(ValueError) as exc:
+        generate_subtarget(single_source_tgt, full_file_name="src/fortran/fake_file.f95")
+    assert "does not match a file src/fortran/fake_file.f95" in str(exc.value)
+
+    class MissingFieldsTarget(Target):
+        alias = "missing_fields_tgt"
         core_fields = (Tags,)
 
-    no_sources_tgt = NoSourcesTgt({Tags.alias: ["demo"]}, address=Address.parse("//:no_sources"))
+    missing_fields_tgt = MissingFieldsTarget(
+        {Tags.alias: ["demo"]}, address=Address("", target_name="missing_fields")
+    )
     with pytest.raises(ValueError) as exc:
-        generate_subtarget(no_sources_tgt, full_file_name="fake.txt")
-    assert "does not support dependencies" in str(exc.value)
+        generate_subtarget(missing_fields_tgt, full_file_name="fake.txt")
+    assert "does not have both a `dependencies` and `sources` field" in str(exc.value)
 
 
 # -----------------------------------------------------------------------------------------------