diff --git a/Cargo.lock b/Cargo.lock
index 6028582..2c743ee 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1992,6 +1992,7 @@ dependencies = [
  "chrono",
  "clap",
  "futures",
+ "http",
  "k8s-openapi",
  "kube",
  "kubert",
@@ -2008,6 +2009,7 @@ dependencies = [
  "tokio",
  "tokio-context",
  "tokio-stream",
+ "tower",
  "tracing",
 ]
 
diff --git a/Cargo.toml b/Cargo.toml
index 78e6c38..3eb8915 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -34,6 +34,8 @@ tokio-stream = "^0.1.18"
 regex = "^1.12.3"
 
 [dev-dependencies]
+http = "1.4.0"
+tower = { version = "0.5.2", features = ["util"] }
 rstest = "^0.26.1"
 once_cell = "^1.21.3"
 chrono = "^0.4.42"
diff --git a/src/controller.rs b/src/controller.rs
index ed03ea4..11065a4 100644
--- a/src/controller.rs
+++ b/src/controller.rs
@@ -215,8 +215,6 @@ async fn reconcile_normally(
     Ok(Action::await_change())
 }
 
-// TODO: Until CDC finalizer is implemented (and even after that if a CDC is deleted using foreground propagation) CDC deletion could cause the cluster to be deleted or unreachable before we have a chance to clean up the resource(s) owned by the ResourceSync, we should be able to detect and handle this scenario gracefully
-// TODO: Immutability on source/target (via CEL?)
 // TODO: If secrets for remote clusters on target and source (when applicable) no longer exist then simply allow the ResourceSync to be deleted by removing the finalizer
 
 async fn reconcile(resource_sync: Arc<ResourceSync>, ctx: Arc<Context>) -> Result<Action> {
@@ -432,21 +430,599 @@ pub async fn run(client: Client) -> Result<()> {
 
 #[cfg(test)]
 mod tests {
+    use super::{
+        reconcile, reconcile_deleted_resource, reconcile_helper, reconcile_normally, Context,
+        RemoteWatcherManager,
+    };
     use super::{
         reconcile_status, sync_failing_transition_time, RESOURCE_SYNC_FAILING_CONDITION,
         RESOURCE_SYNC_SUCCEEDED_REASON,
     };
     use crate::resources::{ResourceSync, ResourceSyncStatus};
+    use crate::test_support::{
+        api_error, discovery_response, resource_sync as sync_fixture, response, MockApi,
+    };
+    use crate::FINALIZER;
     use crate::{Error, Result};
-    use chrono::{TimeDelta, TimeZone};
+    use chrono::TimeZone;
     use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition;
     use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
     use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
     use kube::runtime::controller::Action;
     use once_cell::sync::Lazy;
     use rstest::rstest;
+    use serde_json::json;
+    use std::{sync::Arc, time::Duration};
+
+    fn context(client: kube::Client) -> Arc<Context> {
+        let (remote_watcher_manager, _events) = RemoteWatcherManager::new(client.clone());
+        Arc::new(Context {
+            client,
+            remote_watcher_manager,
+        })
+    }
+
+    async fn stop_watches(ctx: &Context) {
+        tokio::time::timeout(
+            Duration::from_secs(2),
+            ctx.remote_watcher_manager.stop_all(),
+        )
+        .await
+        .expect("watchers cancel and join");
+    }
+
+    const SYNC_PATH: &str =
+        "/apis/sinker.influxdata.io/v1alpha1/namespaces/team-a/resourcesyncs/copy-config";
+    const STATUS_PATH: &str =
+        "/apis/sinker.influxdata.io/v1alpha1/namespaces/team-a/resourcesyncs/copy-config/status";
+    const SOURCE_PATH: &str = "/api/v1/namespaces/team-a/configmaps/source-config";
+    const TARGET_PATH: &str = "/api/v1/namespaces/team-a/configmaps/target-config";
+
+    #[tokio::test]
+    async fn initialization_preserves_finalizers_and_writes_live_success_status() {
+        let mut sync = sync_fixture();
+        sync.metadata.finalizers = Some(vec!["example.com/other".into()]);
+        sync.status = status_with_condition("True");
+        let mut live = sync.clone();
+        live.status = status_with_condition("False");
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            discovery_response("ConfigMap", "configmaps", true),
+            response(200, json!(sync)),
+            response(200, json!(live)),
+            response(200, json!(live)),
+        ]);
+        let result = reconcile(Arc::new(sync), context(mock.client.clone()))
+            .await
+            .expect("initialize sync");
+        assert_eq!(result, Action::requeue(Duration::from_millis(500)));
+        let requests = mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1"),
+            ("PATCH", SYNC_PATH),
+            ("GET", STATUS_PATH),
+            ("PATCH", STATUS_PATH),
+        ]);
+        assert_eq!(
+            requests[2].body(),
+            &json!({"metadata": {"finalizers": ["example.com/other", FINALIZER]}})
+        );
+        assert_eq!(
+            requests[2].headers()["content-type"],
+            "application/merge-patch+json"
+        );
+        let status: ResourceSyncStatus =
+            serde_json::from_value(requests[4].body()["status"].clone()).expect("status patch");
+        let condition = single_condition(Some(status));
+        assert_eq!(condition.last_transition_time, *EPOCH);
+        assert_eq!(condition.observed_generation, Some(7));
+        assert_eq!(condition.status, "False");
+        assert_eq!(condition.message, "Sync succeeded");
+    }
+
+    #[tokio::test]
+    async fn deleted_missing_target_removes_only_our_finalizer_and_skips_status() {
+        let mut sync = sync_fixture();
+        sync.metadata.deletion_timestamp = Some(EPOCH.clone());
+        sync.metadata.finalizers = Some(vec![
+            FINALIZER.into(),
+            "example.com/keep".into(),
+            FINALIZER.into(),
+        ]);
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            discovery_response("ConfigMap", "configmaps", true),
+            api_error(404),
+            response(200, json!(sync)),
+        ]);
+        assert_eq!(
+            reconcile(Arc::new(sync), context(mock.client.clone()))
+                .await
+                .expect("cleanup"),
+            Action::await_change()
+        );
+        let requests = mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1"),
+            ("GET", TARGET_PATH),
+            ("PATCH", SYNC_PATH),
+        ]);
+        assert_eq!(
+            requests[3].body(),
+            &json!({"metadata": {"finalizers": ["example.com/keep"]}})
+        );
+    }
+
+    #[tokio::test]
+    async fn source_failure_is_returned_and_written_to_status() {
+        let mut sync = sync_fixture();
+        sync.metadata.finalizers = Some(vec![FINALIZER.into()]);
+        let mut live = sync.clone();
+        live.status = status_with_condition("True");
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            discovery_response("ConfigMap", "configmaps", true),
+            api_error(404),
+            response(200, json!(live)),
+            response(200, json!(live)),
+        ]);
+        let error = reconcile(Arc::new(sync), context(mock.client.clone()))
+            .await
+            .expect_err("source absent");
+        let message = error.to_string();
+        assert!(
+            matches!(error, Error::ResourceNotFoundError(name, kind, kube::Error::Api(error))
+            if name == "source-config" && kind == "ConfigMap" && error.code == 404)
+        );
+        let requests = mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1"),
+            ("GET", SOURCE_PATH),
+            ("GET", STATUS_PATH),
+            ("PATCH", STATUS_PATH),
+        ]);
+        let condition = &requests[4].body()["status"]["conditions"][0];
+        assert_eq!(condition["status"], "True");
+        assert_eq!(condition["message"], message);
+        assert_eq!(condition["observedGeneration"], 7);
+        assert_eq!(condition["lastTransitionTime"], json!(*EPOCH));
+    }
+
+    #[rstest]
+    #[case::deleting_target_api_failure(true, true, false, true)]
+    #[case::deleting_source_api_failure(true, true, true, true)]
+    #[case::disabled_force_delete(true, false, false, false)]
+    #[case::active_sync(false, true, false, false)]
+    #[tokio::test]
+    async fn force_delete_only_bypasses_api_resolution_for_deleting_syncs(
+        #[case] deleted: bool,
+        #[case] force: bool,
+        #[case] source_failure: bool,
+        #[case] removed: bool,
+    ) {
+        let mut sync = sync_fixture();
+        sync.metadata.deletion_timestamp = deleted.then(|| EPOCH.clone());
+        sync.metadata.finalizers = Some(vec![FINALIZER.into(), "example.com/keep".into()]);
+        sync.metadata.annotations = Some(std::collections::BTreeMap::from([(
+            crate::resources::FORCE_DELETE_ANNOTATION.into(),
+            force.to_string(),
+        )]));
+        let mut responses = vec![];
+        let mut expected = vec![];
+        if source_failure {
+            responses.push(discovery_response("ConfigMap", "configmaps", true));
+            expected.push(("GET", "/api/v1"));
+        }
+        responses.push(api_error(403));
+        expected.push(("GET", "/api/v1"));
+        if removed {
+            responses.push(response(200, json!(sync)));
+            expected.push(("PATCH", SYNC_PATH));
+        }
+        let mock = MockApi::new(responses);
+        let parent_api = sync.api(mock.client.clone());
+        let result = reconcile_helper(
+            Arc::new(sync),
+            context(mock.client.clone()),
+            &"copy-config".into(),
+            &parent_api,
+        )
+        .await;
+        if removed {
+            assert_eq!(result.expect("force cleanup"), Action::await_change());
+        } else {
+            assert!(
+                matches!(result.expect_err("API resolution failure"), Error::KubeError(kube::Error::Api(error)) if error.code == 403)
+            );
+        }
+        let requests = mock.finish(&expected);
+        if removed {
+            assert_eq!(
+                requests.last().expect("finalizer patch").body(),
+                &json!({"metadata": {"finalizers": ["example.com/keep"]}})
+            );
+        }
+    }
+
+    #[rstest]
+    #[case::no_finalizers(None, false, Some("Foreground"))]
+    #[case::empty_finalizers(Some(vec![]), false, Some("Foreground"))]
+    #[case::target_finalizer(Some(vec!["example.com/target"]), false, Some("Background"))]
+    #[case::already_deleting(Some(vec!["example.com/target"]), true, None)]
+    #[tokio::test]
+    async fn target_deletion_waits_for_absence(
+        #[case] finalizers: Option<Vec<&str>>,
+        #[case] deleting: bool,
+        #[case] propagation: Option<&str>,
+    ) {
+        let mut sync = sync_fixture();
+        sync.metadata.finalizers = Some(vec![FINALIZER.into()]);
+        sync.metadata.deletion_timestamp = Some(EPOCH.clone());
+        let target = json!({"apiVersion": "v1", "kind": "ConfigMap", "metadata": {
+            "name": "target-config", "finalizers": finalizers, "deletionTimestamp": deleting.then(|| EPOCH.clone())}});
+        let mut responses = vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            response(200, target.clone()),
+        ];
+        let mut expected = vec![("GET", "/api/v1"), ("GET", TARGET_PATH)];
+        if propagation.is_some() {
+            responses.push(response(200, target));
+            expected.push(("DELETE", TARGET_PATH));
+        }
+        let mock = MockApi::new(responses);
+        let ctx = context(mock.client.clone());
+        let _cancelled =
+            crate::remote_watcher_manager::tests::park_watchers(&ctx.remote_watcher_manager, &sync)
+                .await;
+        let parent = sync.api(mock.client.clone());
+        let target_api = sync
+            .spec
+            .target
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("target API");
+        let result = reconcile_deleted_resource(
+            Arc::new(sync),
+            "copy-config",
+            target_api,
+            &parent,
+            Arc::clone(&ctx),
+        )
+        .await;
+        stop_watches(&ctx).await;
+        assert_eq!(result.expect("request deletion"), Action::await_change());
+        let requests = mock.finish(&expected);
+        if let Some(propagation) = propagation {
+            assert_eq!(requests[2].body()["propagationPolicy"], propagation);
+        }
+    }
+
+    #[rstest]
+    #[case::without_our_finalizer(false, false)]
+    #[case::deletion_disabled(true, true)]
+    #[tokio::test]
+    async fn cleanup_can_skip_target_requests(#[case] finalizer: bool, #[case] disabled: bool) {
+        let mut sync = sync_fixture();
+        sync.metadata.finalizers = Some(if finalizer {
+            vec![FINALIZER.into()]
+        } else {
+            vec!["example.com/other".into()]
+        });
+        sync.metadata.annotations = Some(std::collections::BTreeMap::from([(
+            crate::resources::DISABLE_TARGET_DELETION_ANNOTATION.into(),
+            disabled.to_string(),
+        )]));
+        let mut responses = vec![discovery_response("ConfigMap", "configmaps", true)];
+        let mut expected = vec![("GET", "/api/v1")];
+        if disabled {
+            responses.push(response(200, json!(sync)));
+            expected.push(("PATCH", SYNC_PATH));
+        }
+        let mock = MockApi::new(responses);
+        let parent = sync.api(mock.client.clone());
+        let target = sync
+            .spec
+            .target
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("target API");
+        assert_eq!(
+            reconcile_deleted_resource(
+                Arc::new(sync),
+                "copy-config",
+                target,
+                &parent,
+                context(mock.client.clone())
+            )
+            .await
+            .expect("cleanup"),
+            Action::await_change()
+        );
+        let requests = mock.finish(&expected);
+        if disabled {
+            assert_eq!(requests[1].body(), &json!({"metadata": {"finalizers": []}}));
+        }
+    }
+
+    #[rstest]
+    #[case::whole_resource(false, false)]
+    #[case::mapped_resource(true, false)]
+    #[case::remote_target(false, true)]
+    #[tokio::test]
+    async fn target_apply_uses_forced_field_manager_and_local_ownership(
+        #[case] mapped: bool,
+        #[case] remote: bool,
+    ) {
+        let mut sync = sync_fixture();
+        if mapped {
+            sync.spec.mappings = vec![crate::resources::Mapping {
+                from_field_path: Some("data.original".into()),
+                to_field_path: Some("data.copied".into()),
+            }];
+        }
+        let source = json!({"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "source-config"}, "data": {"original": "value"}});
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            discovery_response("ConfigMap", "configmaps", true),
+            response(200, source.clone()),
+            response(200, source),
+        ]);
+        let source_api = sync
+            .spec
+            .source
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("source API");
+        let target_api = sync
+            .spec
+            .target
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("target API");
+        // API resolution is tested separately; this flag determines target ownership.
+        if remote {
+            sync.spec.target.cluster = Some(Default::default());
+        }
+        let ctx = context(mock.client.clone());
+        let _cancelled =
+            crate::remote_watcher_manager::tests::park_watchers(&ctx.remote_watcher_manager, &sync)
+                .await;
+        let result = reconcile_normally(
+            Arc::new(sync),
+            "copy-config",
+            source_api,
+            target_api,
+            Arc::clone(&ctx),
+        )
+        .await;
+        stop_watches(&ctx).await;
+        assert_eq!(result.expect("apply target"), Action::await_change());
+        let requests = mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1"),
+            ("GET", SOURCE_PATH),
+            ("PATCH", TARGET_PATH),
+        ]);
+        let patch = &requests[3];
+        let query = patch.uri().query().expect("apply parameters");
+        assert!(query.split('&').any(|part| part == "force=true"));
+        assert!(query
+            .split('&')
+            .any(|part| part == "fieldManager=sinker.influxdata.io"));
+        assert_eq!(
+            patch.headers()["content-type"],
+            "application/apply-patch+yaml"
+        );
+        assert_eq!(patch.body()["metadata"]["name"], "target-config");
+        assert_eq!(patch.body()["metadata"]["namespace"], "team-a");
+        assert_eq!(
+            patch.body()["data"],
+            if mapped {
+                json!({"copied": "value"})
+            } else {
+                json!({"original": "value"})
+            }
+        );
+        if remote {
+            assert!(patch.body()["metadata"].get("ownerReferences").is_none());
+        } else {
+            assert_eq!(
+                patch.body()["metadata"]["ownerReferences"],
+                json!([{
+                "apiVersion": "sinker.influxdata.io/v1alpha1", "kind": "ResourceSync", "name": "copy-config",
+                "uid": "sync-uid", "controller": false, "blockOwnerDeletion": true}])
+            );
+        }
+    }
+
+    #[test]
+    fn deletion_errors_still_update_status_and_ignore_unrelated_conditions() {
+        let mut live = status_with_condition("True").expect("status fixture");
+        live.conditions.as_mut().expect("conditions")[0].type_ = "OtherCondition".into();
+        let mut sync = resource_sync(true, None);
+        sync.metadata.generation = Some(9);
+        let before = chrono::Utc::now();
+        let condition = single_condition(reconcile_status(
+            &sync,
+            &Some(live),
+            &Err(Error::NamespaceRequired),
+        ));
+        let after = chrono::Utc::now();
+        assert_eq!(condition.status, "True");
+        assert_eq!(condition.observed_generation, Some(9));
+        assert_eq!(condition.message, "Namespace is required");
+        assert!((before..=after).contains(&condition.last_transition_time.0));
+    }
+
+    #[rstest]
+    #[case::missing_name(true)]
+    #[case::missing_namespace(false)]
+    #[tokio::test]
+    async fn reconciliation_requires_identity_before_api_access(#[case] missing_name: bool) {
+        let mut sync = sync_fixture();
+        let mock = MockApi::new(vec![]);
+        let ctx = context(mock.client.clone());
+        if missing_name {
+            sync.metadata.name = None;
+            assert!(matches!(
+                reconcile(Arc::new(sync), ctx)
+                    .await
+                    .expect_err("name required"),
+                Error::NameRequired
+            ));
+        } else {
+            sync.metadata.namespace = None;
+            let parent = sync.api(mock.client.clone());
+            assert!(matches!(
+                reconcile_helper(Arc::new(sync), ctx, &"copy-config".into(), &parent)
+                    .await
+                    .expect_err("namespace required"),
+                Error::NamespaceRequired
+            ));
+        }
+        mock.finish(&[]);
+    }
+
+    #[rstest]
+    #[case::get_target(false)]
+    #[case::delete_target(true)]
+    #[tokio::test]
+    async fn force_delete_does_not_bypass_target_request_errors(#[case] delete: bool) {
+        let mut sync = sync_fixture();
+        sync.metadata.deletion_timestamp = Some(EPOCH.clone());
+        sync.metadata.finalizers = Some(vec![FINALIZER.into()]);
+        sync.metadata.annotations = Some(std::collections::BTreeMap::from([(
+            crate::resources::FORCE_DELETE_ANNOTATION.into(),
+            "true".into(),
+        )]));
+        let mut responses = vec![discovery_response("ConfigMap", "configmaps", true)];
+        let mut expected = vec![("GET", "/api/v1"), ("GET", TARGET_PATH)];
+        if delete {
+            responses.push(response(
+                200,
+                json!({"metadata": {"name": "target-config"}}),
+            ));
+            expected.push(("DELETE", TARGET_PATH));
+        }
+        responses.push(api_error(403));
+        let mock = MockApi::new(responses);
+        let target = sync
+            .spec
+            .target
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("target API");
+        let parent = sync.api(mock.client.clone());
+        let error = reconcile_deleted_resource(
+            Arc::new(sync),
+            "copy-config",
+            target,
+            &parent,
+            context(mock.client.clone()),
+        )
+        .await
+        .expect_err("target request denied");
+        assert!(matches!(error, Error::KubeError(kube::Error::Api(error)) if error.code == 403));
+        mock.finish(&expected);
+    }
+
+    #[rstest]
+    #[case::missing_owner_uid(false)]
+    #[case::apply_denied(true)]
+    #[tokio::test]
+    async fn target_write_failures_are_returned(#[case] uid_present: bool) {
+        let mut sync = sync_fixture();
+        if !uid_present {
+            sync.metadata.uid = None;
+        }
+        let mut responses = vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            discovery_response("ConfigMap", "configmaps", true),
+            response(200, json!({"data": {"key": "value"}})),
+        ];
+        let mut expected = vec![("GET", "/api/v1"), ("GET", "/api/v1"), ("GET", SOURCE_PATH)];
+        if uid_present {
+            responses.push(api_error(409));
+            expected.push(("PATCH", TARGET_PATH));
+        }
+        let mock = MockApi::new(responses);
+        let ctx = context(mock.client.clone());
+        let source = sync
+            .spec
+            .source
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("source API");
+        let target = sync
+            .spec
+            .target
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("target API");
+        let result = reconcile_normally(
+            Arc::new(sync),
+            "copy-config",
+            source,
+            target,
+            Arc::clone(&ctx),
+        )
+        .await;
+        stop_watches(&ctx).await;
+        let error = result.expect_err("target write failure");
+        if uid_present {
+            assert!(
+                matches!(error, Error::KubeError(kube::Error::Api(error)) if error.code == 409)
+            );
+        } else {
+            assert!(matches!(error, Error::UIDRequired));
+        }
+        mock.finish(&expected);
+    }
+
+    #[rstest]
+    #[case::status_read(false)]
+    #[case::status_write(true)]
+    #[tokio::test]
+    async fn status_api_errors_are_returned(#[case] write: bool) {
+        let sync = sync_fixture();
+        let mut responses = vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            discovery_response("ConfigMap", "configmaps", true),
+            response(200, json!(sync)),
+        ];
+        let mut expected = vec![
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1"),
+            ("PATCH", SYNC_PATH),
+            ("GET", STATUS_PATH),
+        ];
+        if write {
+            responses.push(response(200, json!(sync)));
+            expected.push(("PATCH", STATUS_PATH));
+        }
+        responses.push(api_error(403));
+        let mock = MockApi::new(responses);
+        let error = reconcile(Arc::new(sync), context(mock.client.clone()))
+            .await
+            .expect_err("status request denied");
+        assert!(matches!(error, Error::KubeError(kube::Error::Api(error)) if error.code == 403));
+        mock.finish(&expected);
+    }
+
+    #[tokio::test]
+    async fn reconciliation_errors_retry_after_five_seconds() {
+        let mock = MockApi::new(vec![]);
+        assert_eq!(
+            super::error_policy(
+                Arc::new(sync_fixture()),
+                &Error::NamespaceRequired,
+                context(mock.client.clone())
+            ),
+            Action::requeue(Duration::from_secs(5))
+        );
+        mock.finish(&[]);
+    }
 
-    static NOW: Lazy<Time> = Lazy::new(|| Time(chrono::Utc::now()));
     static EPOCH: Lazy<Time> = Lazy::new(|| Time(chrono::Utc.timestamp_opt(0, 0).unwrap()));
 
     fn status_with_condition(status: &str) -> Option<ResourceSyncStatus> {
@@ -463,23 +1039,26 @@ mod tests {
     }
 
     #[rstest]
-    #[case::none(None, "True", &NOW)]
-    #[case::no_conditions(Some(ResourceSyncStatus::default()), "True", &NOW)]
-    #[case::empty_conditions(Some(ResourceSyncStatus{conditions: Some(vec![])}), "True", &NOW)]
-    #[case::still_failing_keeps_time(status_with_condition("True"), "True", &EPOCH)]
-    #[case::still_succeeding_keeps_time(status_with_condition("False"), "False", &EPOCH)]
-    #[case::failure_after_success_transitions(status_with_condition("False"), "True", &NOW)]
-    #[case::success_after_failure_transitions(status_with_condition("True"), "False", &NOW)]
-    #[tokio::test]
-    async fn test_sync_failing_transition_time(
+    #[case::none(None, "True", None)]
+    #[case::no_conditions(Some(ResourceSyncStatus::default()), "True", None)]
+    #[case::empty_conditions(Some(ResourceSyncStatus{conditions: Some(vec![])}), "True", None)]
+    #[case::still_failing_keeps_time(status_with_condition("True"), "True", Some(&*EPOCH))]
+    #[case::still_succeeding_keeps_time(status_with_condition("False"), "False", Some(&*EPOCH))]
+    #[case::failure_after_success_transitions(status_with_condition("False"), "True", None)]
+    #[case::success_after_failure_transitions(status_with_condition("True"), "False", None)]
+    fn test_sync_failing_transition_time(
         #[case] status: Option<ResourceSyncStatus>,
         #[case] new_status: &str,
-        #[case] expected: &Time,
+        #[case] expected: Option<&Time>,
     ) {
+        let before = chrono::Utc::now();
         let result = sync_failing_transition_time(&status, new_status);
-        let diff = (result.0 - expected.0).abs();
-
-        assert!(diff.le(&TimeDelta::minutes(1)))
+        let after = chrono::Utc::now();
+        if let Some(expected) = expected {
+            assert_eq!(&result, expected);
+        } else {
+            assert!((before..=after).contains(&result.0));
+        }
     }
 
     fn resource_sync(deleted: bool, status: Option<ResourceSyncStatus>) -> ResourceSync {
diff --git a/src/filters.rs b/src/filters.rs
index b17eb7c..1cb7ac2 100644
--- a/src/filters.rs
+++ b/src/filters.rs
@@ -133,4 +133,38 @@ mod tests {
         };
         assert_eq!(resource.was_last_modified_by(&MANAGER), expected);
     }
+
+    #[rstest]
+    #[case::unknown_manager(None, Some(false))]
+    #[case::sinker_manager(Some("sinker.influxdata.io"), Some(true))]
+    #[case::external_manager(Some("external"), Some(false))]
+    fn latest_entry_wins_regardless_of_position(
+        #[case] manager: Option<&str>,
+        #[case] expected: Option<bool>,
+    ) {
+        let latest = Time(
+            chrono::Utc
+                .timestamp_opt(1_700_000_000, 0)
+                .single()
+                .expect("fixed timestamp"),
+        );
+        let old = ManagedFieldsEntry {
+            manager: Some("sinker.influxdata.io".into()),
+            time: Some(EPOCH.clone()),
+            ..Default::default()
+        };
+        let current = ManagedFieldsEntry {
+            manager: manager.map(String::from),
+            time: Some(latest),
+            ..Default::default()
+        };
+        for entries in [vec![old.clone(), current.clone()], vec![current, old]] {
+            let mut resource = crate::test_support::resource_sync();
+            resource.metadata.managed_fields = Some(entries);
+            assert_eq!(
+                resource.was_last_modified_by("sinker.influxdata.io"),
+                expected
+            );
+        }
+    }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 7e28ec3..0942d6a 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -71,3 +71,132 @@ mod remote_watcher_manager;
 mod resource_extensions;
 pub mod resources;
 mod util;
+
+#[cfg(test)]
+mod test_support {
+    use std::collections::VecDeque;
+    use std::sync::{Arc, Mutex};
+
+    use http::{Request, Response};
+    use kube::client::Body;
+    use kube::Client;
+    use serde_json::{json, Value};
+
+    use crate::resources::{ClusterResourceRef, ResourceSync, ResourceSyncSpec, GVKN};
+
+    /// A finite, in-memory API script. Unexpected requests fail and are recorded, and
+    /// `finish` checks that every scripted response was consumed.
+    pub struct MockApi {
+        pub client: Client,
+        responses: Arc<Mutex<VecDeque<Response<Body>>>>,
+        requests: Arc<Mutex<Vec<Request<Value>>>>,
+    }
+
+    impl MockApi {
+        pub fn new(responses: Vec<Response<Body>>) -> Self {
+            let responses = Arc::new(Mutex::new(VecDeque::from(responses)));
+            let requests = Arc::new(Mutex::new(Vec::new()));
+            let service = tower::service_fn({
+                let responses = Arc::clone(&responses);
+                let requests = Arc::clone(&requests);
+                move |request: Request<Body>| {
+                    let responses = Arc::clone(&responses);
+                    let requests = Arc::clone(&requests);
+                    async move {
+                        let (parts, body) = request.into_parts();
+                        let bytes = body.collect_bytes().await?;
+                        let body = if bytes.is_empty() {
+                            Value::Null
+                        } else {
+                            serde_json::from_slice(&bytes).map_err(kube::Error::SerdeError)?
+                        };
+                        requests
+                            .lock()
+                            .expect("request lock poisoned")
+                            .push(Request::from_parts(parts, body));
+                        responses
+                            .lock()
+                            .expect("response lock poisoned")
+                            .pop_front()
+                            .ok_or_else(|| kube::Error::Service("unexpected API request".into()))
+                    }
+                }
+            });
+            Self {
+                client: Client::new(service, "client-default"),
+                responses,
+                requests,
+            }
+        }
+
+        pub fn finish(self, expected: &[(&str, &str)]) -> Vec<Request<Value>> {
+            assert!(self
+                .responses
+                .lock()
+                .expect("response lock poisoned")
+                .is_empty());
+            let requests =
+                std::mem::take(&mut *self.requests.lock().expect("request lock poisoned"));
+            let actual: Vec<_> = requests
+                .iter()
+                .map(|request| (request.method().as_str(), request.uri().path()))
+                .collect();
+            assert_eq!(actual, expected);
+            requests
+        }
+    }
+
+    pub fn response(code: u16, body: Value) -> Response<Body> {
+        Response::builder()
+            .status(code)
+            .header("content-type", "application/json")
+            .body(Body::from(
+                serde_json::to_vec(&body).expect("serialize response"),
+            ))
+            .expect("build response")
+    }
+
+    pub fn api_error(code: u16) -> Response<Body> {
+        response(
+            code,
+            json!({
+                "apiVersion": "v1", "kind": "Status", "status": "Failure",
+                "code": code, "reason": "TestFailure", "message": "scripted API failure"
+            }),
+        )
+    }
+
+    pub fn discovery_response(kind: &str, plural: &str, namespaced: bool) -> Response<Body> {
+        response(
+            200,
+            json!({
+                "apiVersion": "v1", "kind": "APIResourceList", "groupVersion": "v1",
+                "resources": [{"name": plural, "kind": kind, "namespaced": namespaced,
+                    "verbs": ["get", "patch", "delete", "watch"]}]
+            }),
+        )
+    }
+
+    pub fn resource_sync() -> ResourceSync {
+        let reference = |name: &str| ClusterResourceRef {
+            resource_ref: GVKN {
+                api_version: "v1".to_string(),
+                kind: "ConfigMap".to_string(),
+                name: name.to_string(),
+            },
+            cluster: None,
+        };
+        let mut sync = ResourceSync::new(
+            "copy-config",
+            ResourceSyncSpec {
+                source: reference("source-config"),
+                target: reference("target-config"),
+                mappings: vec![],
+            },
+        );
+        sync.metadata.namespace = Some("team-a".to_string());
+        sync.metadata.uid = Some("sync-uid".to_string());
+        sync.metadata.generation = Some(7);
+        sync
+    }
+}
diff --git a/src/mapping.rs b/src/mapping.rs
index e1ab969..4851d4c 100644
--- a/src/mapping.rs
+++ b/src/mapping.rs
@@ -241,6 +241,206 @@ mod tests {
 
     use super::*;
 
+    #[rstest]
+    #[case::whole_source(None, json!({"data": {"key": "value"}}))]
+    #[case::empty_source_path(Some(""), json!({"data": {"key": "value"}}))]
+    #[case::existing_value(Some("data.key"), json!("value"))]
+    #[case::missing_value(Some("data.missing"), json!(null))]
+    fn source_path_selection(#[case] path: Option<&str>, #[case] expected: serde_json::Value) {
+        let source = json!({"data": {"key": "value"}});
+        assert_eq!(
+            find_field_path(source, &path.map(String::from)).expect("select source"),
+            expected
+        );
+    }
+
+    #[rstest]
+    #[case::array_element("spec.items[1]", json!("second"))]
+    #[case::quoted_annotation("metadata.annotations['example.com/key']", json!("annotation"))]
+    fn source_paths_support_jsonpath(#[case] path: &str, #[case] expected: serde_json::Value) {
+        let source = json!({"spec": {"items": ["first", "second"]},
+            "metadata": {"annotations": {"example.com/key": "annotation"}}});
+        assert_eq!(
+            find_field_path(source, &Some(path.into())).expect("select source"),
+            expected
+        );
+    }
+
+    #[test]
+    fn source_path_rejects_multiple_matches() {
+        let error = find_field_path(json!({"items": [1, 2]}), &Some("items[*]".into()))
+            .expect_err("ambiguous selection must fail");
+        assert!(matches!(error, Error::JsonPathExactlyOneValue(path) if path == "$.items[*]"));
+    }
+
+    #[test]
+    fn source_path_rejects_invalid_syntax() {
+        let error =
+            find_field_path(json!({}), &Some("items[".into())).expect_err("invalid JSONPath");
+        assert!(matches!(error, Error::JsonPathError(_)));
+    }
+
+    #[rstest]
+    #[case::merge_siblings(json!({"spec": {"keep": 1}}), "spec.new", json!({"spec": {"keep": 1, "new": 2}}))]
+    #[case::replace_leaf(json!({"spec": {"old": 1}}), "spec", json!({"spec": 2}))]
+    #[case::empty_key(json!({}), "", json!({"": 2}))]
+    #[case::array_syntax_is_literal(json!({}), "items[0]", json!({"items[0]": 2}))]
+    fn destination_paths_use_object_keys(
+        #[case] mut root: serde_json::Value,
+        #[case] path: &str,
+        #[case] expected: serde_json::Value,
+    ) {
+        set_field_path(&mut root, path, json!(2)).expect("set field");
+        assert_eq!(root, expected);
+    }
+
+    #[rstest]
+    #[case::null_root(json!(null), "key")]
+    #[case::array_root(json!([]), "key")]
+    #[case::scalar_root(json!(false), "key")]
+    #[case::scalar_intermediate(json!({"spec": "occupied"}), "spec.key")]
+    #[case::array_intermediate(json!({"spec": []}), "spec.key")]
+    #[case::null_intermediate(json!({"spec": null}), "spec.key")]
+    fn destination_paths_reject_non_objects(
+        #[case] mut root: serde_json::Value,
+        #[case] path: &str,
+    ) {
+        let original = root.clone();
+        let error = set_field_path(&mut root, path, json!("leaf")).expect_err("object required");
+        assert!(matches!(error, AddToPathError::ObjectRequired(value) if value == original));
+        assert_eq!(root, original);
+    }
+
+    #[rstest]
+    #[case::missing_api_version(json!({"kind": "ConfigMap"}), "apiVersion")]
+    #[case::non_string_api_version(json!({"apiVersion": 1, "kind": "ConfigMap"}), "apiVersion")]
+    #[case::missing_kind(json!({"apiVersion": "v1"}), "kind")]
+    #[case::non_string_kind(json!({"apiVersion": "v1", "kind": []}), "kind")]
+    fn inner_resource_requires_type_metadata(
+        #[case] subtree: serde_json::Value,
+        #[case] field: &str,
+    ) {
+        let error = get_ar_from_subtree(&subtree).expect_err("malformed inner resource");
+        assert!(matches!(error, Error::MalformedInnerResource(message) if message.contains(field)));
+    }
+
+    #[rstest]
+    #[case::namespaced(Some("destination"))]
+    #[case::cluster_scoped(None)]
+    fn clone_preserves_payload_and_user_metadata_only(#[case] namespace: Option<&str>) {
+        let source: DynamicObject = serde_json::from_value(json!({
+            "apiVersion": "v1", "kind": "ConfigMap",
+            "metadata": {
+                "name": "original", "namespace": "source", "uid": "source-uid",
+                "resourceVersion": "123", "generation": 4,
+                "creationTimestamp": "2024-01-01T00:00:00Z", "finalizers": ["other/finalizer"],
+                "ownerReferences": [{"apiVersion": "v1", "kind": "Pod", "name": "owner", "uid": "owner-uid"}],
+                "managedFields": [{"manager": "someone"}],
+                "labels": {"app": "demo"},
+                "annotations": {"example.com/keep": "yes", "kubectl.kubernetes.io/last-applied-configuration": "discard"}
+            },
+            "data": {"nested": [1, null, false]}, "status": {"ready": true}
+        })).expect("source fixture");
+        let original = json!(source);
+        let target_ref = GVKN {
+            api_version: "example.com/v2".into(),
+            kind: "Destination".into(),
+            name: "copy".into(),
+        };
+        let ar = ApiResource::from_gvk(&GroupVersionKind::gvk("example.com", "v2", "Destination"));
+        let target = clone_resource(&source, &target_ref, namespace, &ar).expect("clone source");
+        let mut expected = json!({"apiVersion": "example.com/v2", "kind": "Destination",
+            "metadata": {"name": "copy", "labels": {"app": "demo"}, "annotations": {"example.com/keep": "yes"}},
+            "data": {"nested": [1, null, false]}, "status": {"ready": true}});
+        if let Some(namespace) = namespace {
+            expected["metadata"]["namespace"] = json!(namespace);
+        }
+        assert_eq!(json!(target), expected);
+        assert_eq!(json!(source), original);
+    }
+
+    fn mapped(
+        source: serde_json::Value,
+        paths: &[(Option<&str>, Option<&str>)],
+    ) -> crate::Result<DynamicObject> {
+        let source: DynamicObject = serde_json::from_value(source).expect("source fixture");
+        let mut sync = crate::test_support::resource_sync();
+        sync.spec.mappings = paths
+            .iter()
+            .map(|(from, to)| Mapping {
+                from_field_path: from.map(String::from),
+                to_field_path: to.map(String::from),
+            })
+            .collect();
+        let ar = ApiResource::from_gvk(&GroupVersionKind::gvk("", "v1", "ConfigMap"));
+        apply_mappings(&source, &sync.spec.target.resource_ref, None, &ar, &sync)
+    }
+
+    #[test]
+    fn empty_mapping_entry_is_an_error() {
+        let error =
+            mapped(json!({"data": {"key": "value"}}), &[(None, None)]).expect_err("empty mapping");
+        assert!(matches!(error, Error::MappingEmpty));
+    }
+
+    #[test]
+    fn mappings_are_ordered_and_missing_sources_become_null() {
+        let target = mapped(
+            json!({"data": {"first": "one", "second": "two"}}),
+            &[
+                (Some("data.first"), Some("data.value")),
+                (Some("data.second"), Some("data.value")),
+                (Some("data.missing"), Some("data.absent")),
+            ],
+        )
+        .expect("apply ordered mappings");
+        assert_eq!(
+            json!(target),
+            json!({"apiVersion": "v1", "kind": "ConfigMap",
+            "metadata": {"name": "target-config"}, "data": {"value": "two", "absent": null}})
+        );
+    }
+
+    #[test]
+    fn root_replacement_discards_earlier_mappings_and_allows_later_mappings() {
+        let target = mapped(json!({"spec": {"apiVersion": "v1", "kind": "ConfigMap",
+            "metadata": {"labels": {"app": "embedded"}}, "data": {"embedded": "value"}}, "extra": "after"}), &[
+            (Some("extra"), Some("data.discarded")),
+            (Some("spec"), None),
+            (Some("extra"), Some("data.kept")),
+        ]).expect("replace root");
+        assert_eq!(
+            json!(target),
+            json!({"apiVersion": "v1", "kind": "ConfigMap",
+            "metadata": {"name": "target-config", "labels": {"app": "embedded"}},
+            "data": {"embedded": "value", "kept": "after"}})
+        );
+        assert!(target.data.get("metadata").is_none());
+        assert!(target.data.get("apiVersion").is_none());
+        assert!(target.data.get("kind").is_none());
+    }
+
+    #[test]
+    fn mapping_errors_propagate_from_destination_and_metadata_conversion() {
+        let error = mapped(
+            json!({"value": 42}),
+            &[
+                (Some("value"), Some("data")),
+                (Some("value"), Some("data.child")),
+            ],
+        )
+        .expect_err("cannot traverse scalar");
+        assert!(
+            matches!(error, Error::AddToPathError(AddToPathError::ObjectRequired(value)) if value == json!({"data": 42}))
+        );
+        let error = mapped(
+            json!({"value": 42}),
+            &[(Some("value"), Some("metadata.labels.app"))],
+        )
+        .expect_err("metadata labels must be strings");
+        assert!(matches!(error, Error::SerializationError(error) if error.is_data()));
+    }
+
     #[rstest]
     #[case("status", r#"{"spec":{},"status":"demo"}"#)]
     #[case("status.foo", r#"{"spec":{},"status":{"keep":1,"foo":"demo"}}"#)]
diff --git a/src/remote_watcher.rs b/src/remote_watcher.rs
index d314984..5ed669e 100644
--- a/src/remote_watcher.rs
+++ b/src/remote_watcher.rs
@@ -214,3 +214,313 @@ impl RemoteWatcher {
         Ok(resource_version)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_support::{api_error, discovery_response, resource_sync, response, MockApi};
+    use http::Response;
+    use kube::client::Body;
+    use rstest::rstest;
+    use serde_json::{json, Value};
+    use tokio::sync::mpsc;
+    use tokio::time::timeout;
+
+    fn watcher(
+        client: Client,
+    ) -> (
+        RemoteWatcher,
+        mpsc::UnboundedReceiver<ObjectRef<ResourceSync>>,
+    ) {
+        let sync = resource_sync();
+        let key = RemoteWatcherKey {
+            object: sync.spec.source.clone(),
+            resource_sync: ObjectRef::from_obj(&sync),
+        };
+        let (sender, receiver) = mpsc::unbounded_channel();
+        (RemoteWatcher::new(key, sender, client), receiver)
+    }
+
+    fn watch_response(events: Vec<Value>) -> Response<Body> {
+        let mut bytes = Vec::new();
+        for event in events {
+            serde_json::to_writer(&mut bytes, &event).expect("serialize watch event");
+            bytes.push(b'\n');
+        }
+        Response::builder()
+            .status(200)
+            .body(Body::from(bytes))
+            .expect("watch response")
+    }
+
+    fn object_event(event: &str, manager: Option<&str>, rv: Option<&str>) -> Value {
+        let managed_fields = manager
+            .map(|manager| vec![json!({"manager": manager, "time": "2024-01-01T00:00:00Z"})]);
+        json!({"type": event, "object": {"apiVersion": "v1", "kind": "ConfigMap",
+            "metadata": {"name": "source-config", "resourceVersion": rv, "managedFields": managed_fields}}})
+    }
+
+    #[rstest]
+    #[case::added_external(object_event("ADDED", Some("external"), Some("11")), "11", true)]
+    #[case::added_sinker(
+        object_event("ADDED", Some("sinker.influxdata.io"), Some("11")),
+        "11",
+        false
+    )]
+    #[case::modified_external(object_event("MODIFIED", Some("external"), Some("12")), "12", true)]
+    #[case::modified_sinker(
+        object_event("MODIFIED", Some("sinker.influxdata.io"), Some("12")),
+        "12",
+        false
+    )]
+    #[case::unknown_ownership(object_event("MODIFIED", None, Some("13")), "13", true)]
+    #[case::deleted_sinker(
+        object_event("DELETED", Some("sinker.influxdata.io"), Some("14")),
+        "14",
+        true
+    )]
+    #[case::bookmark(json!({"type": "BOOKMARK", "object": {"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"resourceVersion": "15"}}}), "15", false)]
+    #[case::expired_version(json!({"type": "ERROR", "object": {"code": 410, "reason": "Expired", "message": "too old", "status": "Failure"}}), "0", false)]
+    #[case::api_error(json!({"type": "ERROR", "object": {"code": 500, "reason": "InternalError", "message": "retry", "status": "Failure"}}), "10", true)]
+    #[tokio::test]
+    async fn watch_events_advance_versions_and_reconcile_when_needed(
+        #[case] event: Value,
+        #[case] expected_rv: &str,
+        #[case] reconcile: bool,
+    ) {
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            watch_response(vec![event]),
+        ]);
+        let (watcher, mut receiver) = watcher(mock.client.clone());
+        let api = watcher
+            .key
+            .object
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("API discovery");
+        let rv = timeout(
+            Duration::from_secs(3),
+            watcher.listen(
+                &api,
+                "10".into(),
+                &WatchParams::default(),
+                &mut DefaultBackoff::default(),
+            ),
+        )
+        .await
+        .expect("bounded watch stream")
+        .expect("read events");
+        assert_eq!(rv, expected_rv);
+        if reconcile {
+            assert_eq!(
+                receiver.try_recv().expect("reconcile event"),
+                watcher.key.resource_sync
+            );
+        }
+        assert!(matches!(
+            receiver.try_recv(),
+            Err(mpsc::error::TryRecvError::Empty)
+        ));
+        mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1/namespaces/team-a/configmaps"),
+        ]);
+    }
+
+    #[rstest]
+    #[case::added("ADDED")]
+    #[case::modified("MODIFIED")]
+    #[case::deleted("DELETED")]
+    #[tokio::test]
+    async fn object_events_require_resource_version(#[case] event: &str) {
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            watch_response(vec![object_event(event, None, None)]),
+        ]);
+        let (watcher, mut receiver) = watcher(mock.client.clone());
+        let api = watcher
+            .key
+            .object
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("API discovery");
+        let error = watcher
+            .listen(
+                &api,
+                "10".into(),
+                &WatchParams::default(),
+                &mut DefaultBackoff::default(),
+            )
+            .await
+            .expect_err("missing event version");
+        assert!(matches!(error, Error::ResourceVersionRequired));
+        assert!(matches!(
+            receiver.try_recv(),
+            Err(mpsc::error::TryRecvError::Empty)
+        ));
+        mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1/namespaces/team-a/configmaps"),
+        ]);
+    }
+
+    #[tokio::test]
+    async fn start_reconciles_then_reconnects_from_last_event_with_name_selector() {
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            response(
+                200,
+                json!({"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "source-config", "resourceVersion": "10"}}),
+            ),
+            watch_response(vec![object_event("MODIFIED", Some("external"), Some("11"))]),
+            api_error(403),
+        ]);
+        let (watcher, mut receiver) = watcher(mock.client.clone());
+        let error = timeout(
+            Duration::from_secs(2),
+            watcher.start(&mut DefaultBackoff::default()),
+        )
+        .await
+        .expect("bounded reconnect")
+        .expect_err("second watch denied");
+        assert!(matches!(error, Error::KubeError(kube::Error::Api(error)) if error.code == 403));
+        for _ in 0..2 {
+            assert_eq!(
+                receiver
+                    .try_recv()
+                    .expect("initial and external reconciles"),
+                watcher.key.resource_sync
+            );
+        }
+        assert!(matches!(
+            receiver.try_recv(),
+            Err(mpsc::error::TryRecvError::Empty)
+        ));
+        let requests = mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1/namespaces/team-a/configmaps/source-config"),
+            ("GET", "/api/v1/namespaces/team-a/configmaps"),
+            ("GET", "/api/v1/namespaces/team-a/configmaps"),
+        ]);
+        for (request, rv) in [
+            (&requests[2], "resourceVersion=10"),
+            (&requests[3], "resourceVersion=11"),
+        ] {
+            let parameters: Vec<_> = request
+                .uri()
+                .query()
+                .expect("watch query")
+                .split('&')
+                .collect();
+            assert!(parameters.contains(&"watch=true"));
+            assert!(parameters.contains(&"fieldSelector=metadata.name%3Dsource-config"));
+            assert!(parameters.contains(&rv));
+        }
+    }
+
+    #[tokio::test]
+    async fn start_requires_namespace_before_api_access() {
+        let mock = MockApi::new(vec![]);
+        let (mut watcher, _receiver) = watcher(mock.client.clone());
+        watcher.key.resource_sync.namespace = None;
+        assert!(matches!(
+            watcher
+                .start(&mut DefaultBackoff::default())
+                .await
+                .expect_err("namespace required"),
+            Error::NamespaceRequired
+        ));
+        mock.finish(&[]);
+    }
+
+    #[tokio::test]
+    async fn start_requires_initial_resource_version_before_sending_reconcile() {
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            response(200, json!({"metadata": {"name": "source-config"}})),
+        ]);
+        let (watcher, mut receiver) = watcher(mock.client.clone());
+        assert!(matches!(
+            watcher
+                .start(&mut DefaultBackoff::default())
+                .await
+                .expect_err("resource version required"),
+            Error::ResourceVersionRequired
+        ));
+        assert!(matches!(
+            receiver.try_recv(),
+            Err(mpsc::error::TryRecvError::Empty)
+        ));
+        mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1/namespaces/team-a/configmaps/source-config"),
+        ]);
+    }
+
+    #[tokio::test]
+    async fn failed_start_requests_reconciliation_and_can_be_cancelled() {
+        let mock = MockApi::new(vec![]);
+        let (mut watcher, mut receiver) = watcher(mock.client.clone());
+        watcher.key.resource_sync.namespace = None;
+        let expected = watcher.key.resource_sync.clone();
+        let (ctx, handle) = Context::new();
+        let task = tokio::spawn(watcher.run(ctx));
+        let event = timeout(Duration::from_secs(2), receiver.recv()).await;
+        handle.cancel();
+        timeout(Duration::from_secs(3), task)
+            .await
+            .expect("join cancelled watch")
+            .expect("watch task did not panic");
+        assert_eq!(event.expect("failure triggers reconcile"), Some(expected));
+        mock.finish(&[]);
+    }
+
+    #[tokio::test]
+    async fn sending_to_closed_reconcile_channel_is_nonfatal() {
+        let mock = MockApi::new(vec![]);
+        let (watcher, receiver) = watcher(mock.client.clone());
+        drop(receiver);
+        watcher.send_reconcile_on_success(&mut DefaultBackoff::default());
+        mock.finish(&[]);
+    }
+
+    #[tokio::test]
+    async fn invalid_watch_json_returns_a_decode_error() {
+        let mock = MockApi::new(vec![
+            discovery_response("ConfigMap", "configmaps", true),
+            Response::builder()
+                .status(200)
+                .body(Body::from(b"not-json\n".to_vec()))
+                .expect("malformed stream"),
+        ]);
+        let (watcher, mut receiver) = watcher(mock.client.clone());
+        let api = watcher
+            .key
+            .object
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("API discovery");
+        let error = watcher
+            .listen(
+                &api,
+                "10".into(),
+                &WatchParams::default(),
+                &mut DefaultBackoff::default(),
+            )
+            .await
+            .expect_err("decode failure");
+        assert!(matches!(
+            error,
+            Error::KubeError(kube::Error::SerdeError(_))
+        ));
+        assert!(matches!(
+            receiver.try_recv(),
+            Err(mpsc::error::TryRecvError::Empty)
+        ));
+        mock.finish(&[
+            ("GET", "/api/v1"),
+            ("GET", "/api/v1/namespaces/team-a/configmaps"),
+        ]);
+    }
+}
diff --git a/src/remote_watcher_manager.rs b/src/remote_watcher_manager.rs
index 96f5829..f8d1a47 100644
--- a/src/remote_watcher_manager.rs
+++ b/src/remote_watcher_manager.rs
@@ -82,3 +82,181 @@ impl RemoteWatcherManager {
         }
     }
 }
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use super::*;
+    use crate::test_support::{resource_sync, MockApi};
+    use std::time::Duration;
+    use tokio::time::timeout;
+
+    // Keep controller request tests independent of background discovery while still
+    // exercising the manager's existing-watch and cancellation paths.
+    pub async fn park_watchers(
+        manager: &RemoteWatcherManager,
+        sync: &ResourceSync,
+    ) -> mpsc::UnboundedReceiver<RemoteWatcherKey> {
+        let (sender, receiver) = mpsc::unbounded_channel();
+        for object in [&sync.spec.source, &sync.spec.target] {
+            let key = RemoteWatcherKey {
+                object: object.clone(),
+                resource_sync: ObjectRef::from_obj(sync),
+            };
+            let (mut ctx, handle) = Context::new();
+            let sender = sender.clone();
+            let task_key = key.clone();
+            let task = tokio::spawn(async move {
+                ctx.done().await;
+                sender
+                    .send(task_key)
+                    .expect("cancellation receiver remains open");
+            });
+            assert!(manager
+                .watchers
+                .lock()
+                .await
+                .insert(key, (handle, task))
+                .is_none());
+        }
+        receiver
+    }
+
+    #[tokio::test]
+    async fn removal_cancels_and_joins_only_the_requested_watcher() {
+        let mock = MockApi::new(vec![]);
+        let (manager, _events) = RemoteWatcherManager::new(mock.client.clone());
+        let sync = resource_sync();
+        let mut cancelled = park_watchers(&manager, &sync).await;
+        let source = RemoteWatcherKey {
+            object: sync.spec.source.clone(),
+            resource_sync: ObjectRef::from_obj(&sync),
+        };
+        let target = RemoteWatcherKey {
+            object: sync.spec.target.clone(),
+            resource_sync: ObjectRef::from_obj(&sync),
+        };
+        // A duplicate addition must keep the existing task, without starting API discovery.
+        manager.add_if_not_exists(&source).await;
+        timeout(
+            Duration::from_secs(2),
+            manager.stop_and_remove_if_exists(&source),
+        )
+        .await
+        .expect("stop source");
+        assert_eq!(
+            cancelled
+                .try_recv()
+                .expect("source joined after cancellation"),
+            source
+        );
+        assert!(matches!(
+            cancelled.try_recv(),
+            Err(mpsc::error::TryRecvError::Empty)
+        ));
+        {
+            let watchers = manager.watchers.lock().await;
+            assert_eq!(watchers.len(), 1);
+            assert!(watchers.contains_key(&target));
+        }
+        manager.stop_and_remove_if_exists(&source).await;
+        timeout(Duration::from_secs(2), manager.stop_all())
+            .await
+            .expect("stop remaining watcher");
+        assert_eq!(
+            cancelled
+                .try_recv()
+                .expect("target joined after cancellation"),
+            target
+        );
+        assert!(manager.watchers.lock().await.is_empty());
+        assert!(matches!(
+            cancelled.try_recv(),
+            Err(mpsc::error::TryRecvError::Disconnected)
+        ));
+        manager.stop_all().await;
+        mock.finish(&[]);
+    }
+
+    #[tokio::test]
+    async fn stop_all_cancels_and_joins_every_owner_of_the_same_resource() {
+        let mock = MockApi::new(vec![]);
+        let (manager, _events) = RemoteWatcherManager::new(mock.client.clone());
+        let first = resource_sync();
+        let mut second = first.clone();
+        second.metadata.name = Some("other-sync".into());
+        let mut first_cancelled = park_watchers(&manager, &first).await;
+        let mut second_cancelled = park_watchers(&manager, &second).await;
+        assert_eq!(manager.watchers.lock().await.len(), 4);
+        timeout(Duration::from_secs(2), manager.stop_all())
+            .await
+            .expect("stop all watchers");
+        assert!(manager.watchers.lock().await.is_empty());
+        for receiver in [&mut first_cancelled, &mut second_cancelled] {
+            let keys = [
+                receiver.try_recv().expect("first cancellation"),
+                receiver.try_recv().expect("second cancellation"),
+            ];
+            assert_ne!(keys[0].object, keys[1].object);
+            assert!(matches!(
+                receiver.try_recv(),
+                Err(mpsc::error::TryRecvError::Disconnected)
+            ));
+        }
+        mock.finish(&[]);
+    }
+
+    #[tokio::test]
+    async fn new_watcher_can_be_cancelled_before_it_starts_network_work() {
+        let mock = MockApi::new(vec![]);
+        let (manager, _events) = RemoteWatcherManager::new(mock.client.clone());
+        let sync = resource_sync();
+        let key = RemoteWatcherKey {
+            object: sync.spec.source.clone(),
+            resource_sync: ObjectRef::from_obj(&sync),
+        };
+        manager.add_if_not_exists(&key).await;
+        manager.add_if_not_exists(&key).await;
+        assert_eq!(manager.watchers.lock().await.len(), 1);
+        timeout(
+            Duration::from_secs(2),
+            manager.stop_and_remove_if_exists(&key),
+        )
+        .await
+        .expect("cancel new watcher");
+        assert!(manager.watchers.lock().await.is_empty());
+        mock.finish(&[]);
+    }
+
+    #[tokio::test]
+    async fn cleanup_removes_aborted_tasks_and_continues() {
+        let mock = MockApi::new(vec![]);
+        let (manager, _events) = RemoteWatcherManager::new(mock.client.clone());
+        let sync = resource_sync();
+        let mut cancelled = park_watchers(&manager, &sync).await;
+        let source = RemoteWatcherKey {
+            object: sync.spec.source.clone(),
+            resource_sync: ObjectRef::from_obj(&sync),
+        };
+        manager
+            .watchers
+            .lock()
+            .await
+            .get(&source)
+            .expect("source task")
+            .1
+            .abort();
+        timeout(Duration::from_secs(2), manager.stop_all())
+            .await
+            .expect("join aborted and live tasks");
+        assert!(manager.watchers.lock().await.is_empty());
+        assert_eq!(
+            cancelled.try_recv().expect("live target cancelled").object,
+            sync.spec.target
+        );
+        assert!(matches!(
+            cancelled.try_recv(),
+            Err(mpsc::error::TryRecvError::Disconnected)
+        ));
+        mock.finish(&[]);
+    }
+}
diff --git a/src/resource_extensions.rs b/src/resource_extensions.rs
index b473a3b..cf0d222 100644
--- a/src/resource_extensions.rs
+++ b/src/resource_extensions.rs
@@ -220,12 +220,232 @@ impl ClusterResourceRef {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_support::{api_error, discovery_response, resource_sync, response, MockApi};
     use futures::future::join_all;
     use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
     use rand::{distr::Alphanumeric, rngs::StdRng, Rng, SeedableRng};
     use rstest::rstest;
+    use serde_json::json;
     use std::collections::BTreeMap;
 
+    #[rstest]
+    #[case::absent(None, false)]
+    #[case::empty(Some(vec![]), false)]
+    #[case::unrelated(Some(vec!["example.com/other"]), false)]
+    #[case::sinker(Some(vec!["example.com/other", FINALIZER]), true)]
+    fn finalizers_are_detected_and_cloned(
+        #[case] finalizers: Option<Vec<&str>>,
+        #[case] expected: bool,
+    ) {
+        let mut sync = resource_sync();
+        sync.metadata.finalizers =
+            finalizers.map(|values| values.into_iter().map(String::from).collect());
+        assert_eq!(sync.has_target_finalizer(), expected);
+        let original = sync.metadata.finalizers.clone();
+        let mut cloned = sync.finalizers_clone_or_empty();
+        assert_eq!(cloned, original.clone().unwrap_or_default());
+        cloned.push("new/finalizer".into());
+        assert_eq!(sync.metadata.finalizers, original);
+    }
+
+    #[rstest]
+    #[case::unanchored("team", "prefix-team-suffix", true)]
+    #[case::anchored("^team$", "prefix-team-suffix", false)]
+    #[case::empty_pattern("", "team-a", true)]
+    #[case::alternatives("^(team-a|team-b)$", "team-b", true)]
+    fn namespace_annotation_uses_regex_matching(
+        #[case] pattern: &str,
+        #[case] namespace: &str,
+        #[case] allowed: bool,
+    ) {
+        let result =
+            verify_kubeconfig_secret_access(namespace, &secret_with_annotation(Some(pattern)));
+        if allowed {
+            result.expect("matching namespace authorized");
+        } else {
+            assert!(matches!(
+                result.expect_err("nonmatching namespace denied"),
+                UnauthorizedKubeconfigAccess()
+            ));
+        }
+    }
+
+    #[tokio::test]
+    async fn local_client_and_resource_sync_api_do_not_require_discovery() {
+        let mock = MockApi::new(vec![]);
+        let client = cluster_client(None, "team-a", mock.client.clone())
+            .await
+            .expect("local client");
+        assert_eq!(client.default_namespace(), "client-default");
+        let mut sync = resource_sync();
+        assert_eq!(
+            sync.api(client.clone()).resource_url(),
+            "/apis/sinker.influxdata.io/v1alpha1/namespaces/team-a/resourcesyncs"
+        );
+        sync.metadata.namespace = None;
+        assert_eq!(
+            sync.api(client).resource_url(),
+            "/apis/sinker.influxdata.io/v1alpha1/resourcesyncs"
+        );
+        mock.finish(&[]);
+    }
+
+    #[rstest]
+    #[case::namespaced(
+        "ConfigMap",
+        "configmaps",
+        true,
+        Some("team-a"),
+        "/api/v1/namespaces/team-a/configmaps"
+    )]
+    #[case::cluster_scoped("Namespace", "namespaces", false, None, "/api/v1/namespaces")]
+    #[tokio::test]
+    async fn local_api_uses_discovered_scope(
+        #[case] kind: &str,
+        #[case] plural: &str,
+        #[case] namespaced: bool,
+        #[case] namespace: Option<&str>,
+        #[case] url: &str,
+    ) {
+        let mock = MockApi::new(vec![discovery_response(kind, plural, namespaced)]);
+        let mut reference = resource_sync().spec.source;
+        reference.resource_ref.kind = kind.into();
+        let api = reference
+            .api_for(mock.client.clone(), "team-a")
+            .await
+            .expect("discover local API");
+        assert_eq!(api.namespace.as_deref(), namespace);
+        assert_eq!(api.ar.kind, kind);
+        assert_eq!(api.resource_url(), url);
+        mock.finish(&[("GET", "/api/v1")]);
+    }
+
+    fn cluster_ref(namespace: Option<&str>) -> ClusterRef {
+        use crate::resources::{KubeConfig, SecretRef};
+        ClusterRef {
+            namespace: Some("workloads".into()),
+            kube_config: KubeConfig {
+                secret_ref: SecretRef {
+                    name: "credentials".into(),
+                    namespace: namespace.map(String::from),
+                    key: "config".into(),
+                },
+            },
+        }
+    }
+
+    #[rstest]
+    #[case::implicit_local(None, "/api/v1/namespaces/team-a/secrets/credentials", false)]
+    #[case::explicit_local(Some("team-a"), "/api/v1/namespaces/team-a/secrets/credentials", false)]
+    #[case::cross_namespace(
+        Some("credential-store"),
+        "/api/v1/namespaces/credential-store/secrets/credentials",
+        true
+    )]
+    #[tokio::test]
+    async fn secret_read_errors_hide_cross_namespace_details(
+        #[case] namespace: Option<&str>,
+        #[case] path: &str,
+        #[case] unauthorized: bool,
+    ) {
+        let mock = MockApi::new(vec![api_error(403)]);
+        let error = cluster_client(Some(&cluster_ref(namespace)), "team-a", mock.client.clone())
+            .await
+            .map(|_| ())
+            .expect_err("secret read fails");
+        if unauthorized {
+            assert!(matches!(error, UnauthorizedKubeconfigAccess()));
+        } else {
+            assert!(
+                matches!(error, Error::KubeError(kube::Error::Api(error)) if error.code == 403)
+            );
+        }
+        mock.finish(&[("GET", path)]);
+    }
+
+    #[rstest]
+    #[case::same_namespace_without_annotation(None, None, true)]
+    #[case::same_namespace_ignores_denial(None, Some("^denied$"), true)]
+    #[case::cross_namespace_missing_annotation(Some("credential-store"), None, false)]
+    #[case::cross_namespace_denied(Some("credential-store"), Some("^denied$"), false)]
+    #[case::cross_namespace_allowed(Some("credential-store"), Some("^team-a$"), true)]
+    #[tokio::test]
+    async fn secret_authorization_precedes_key_lookup(
+        #[case] namespace: Option<&str>,
+        #[case] annotation: Option<&str>,
+        #[case] allowed: bool,
+    ) {
+        let mut secret = secret_with_annotation(annotation);
+        secret.data = Some(BTreeMap::new());
+        let mock = MockApi::new(vec![response(200, json!(secret))]);
+        let error = cluster_client(Some(&cluster_ref(namespace)), "team-a", mock.client.clone())
+            .await
+            .map(|_| ())
+            .expect_err("missing key or authorization");
+        if allowed {
+            assert!(matches!(error, Error::MissingKeyError(key, name, ns)
+                if key == "config" && name == "credentials" && ns == namespace.unwrap_or("team-a")));
+        } else {
+            assert!(matches!(error, UnauthorizedKubeconfigAccess()));
+        }
+        let path = format!(
+            "/api/v1/namespaces/{}/secrets/credentials",
+            namespace.unwrap_or("team-a")
+        );
+        mock.finish(&[("GET", &path)]);
+    }
+
+    #[rstest]
+    #[case::invalid_utf8(vec![0xff], "utf8")]
+    #[case::invalid_yaml(b"[unterminated".to_vec(), "yaml")]
+    #[case::missing_context(b"apiVersion: v1\nkind: Config\nclusters: []\ncontexts: []\nusers: []\n".to_vec(), "context")]
+    #[tokio::test]
+    async fn invalid_kubeconfigs_return_typed_errors(
+        #[case] bytes: Vec<u8>,
+        #[case] failure: &str,
+    ) {
+        let secret = Secret {
+            data: Some(BTreeMap::from([(
+                "config".into(),
+                k8s_openapi::ByteString(bytes),
+            )])),
+            ..Default::default()
+        };
+        let mock = MockApi::new(vec![response(200, json!(secret))]);
+        let error = cluster_client(Some(&cluster_ref(None)), "team-a", mock.client.clone())
+            .await
+            .map(|_| ())
+            .expect_err("invalid kubeconfig");
+        match failure {
+            "utf8" => assert!(
+                matches!(error, Error::KubeconfigUtf8Error(error) if error.valid_up_to() == 0)
+            ),
+            "yaml" | "context" => assert!(matches!(error, Error::KubeconfigError(_))),
+            _ => unreachable!(),
+        }
+        mock.finish(&[("GET", "/api/v1/namespaces/team-a/secrets/credentials")]);
+    }
+
+    #[test]
+    fn watcher_keys_distinguish_owners_and_resource_connections() {
+        let sync = resource_sync();
+        let reference = sync.spec.source.clone();
+        let key = reference.remote_watcher_key(&sync);
+        assert_eq!(key.object, reference);
+        assert_eq!(key.resource_sync.name, "copy-config");
+        assert_eq!(key.resource_sync.namespace.as_deref(), Some("team-a"));
+        let mut other_sync = sync.clone();
+        other_sync.metadata.namespace = Some("team-b".into());
+        assert_ne!(key, reference.remote_watcher_key(&other_sync));
+        other_sync = sync.clone();
+        other_sync.metadata.name = Some("other-sync".into());
+        assert_ne!(key, reference.remote_watcher_key(&other_sync));
+        let mut remote = reference.clone();
+        remote.cluster = Some(cluster_ref(None));
+        assert_ne!(key, remote.remote_watcher_key(&sync));
+        assert_ne!(key, sync.spec.target.remote_watcher_key(&sync));
+    }
+
     fn secret_with_annotation(value: Option<&str>) -> Secret {
         Secret {
             metadata: ObjectMeta {
@@ -326,8 +546,7 @@ mod tests {
                 };
                 let ns = if expect_ok {
                     let part: String = (0..2)
-                        .map(|_| rng.sample(Alphanumeric) as char)
-                        .map(|c| c.to_ascii_lowercase())
+                        .map(|_| char::from(rng.random_range(b'a'..=b'z')))
                         .collect();
                     format!("ok-{}", part)
                 } else {
@@ -342,7 +561,7 @@ mod tests {
             .map(|(ns, expect_ok, sec)| {
                 tokio::spawn(async move {
                     let res = verify_kubeconfig_secret_access(&ns, &sec);
-                    (expect_ok, res)
+                    (ns, expect_ok, res)
                 })
             })
             .collect();
@@ -350,9 +569,12 @@ mod tests {
         let outcomes = join_all(handles).await;
 
         for outcome in outcomes {
-            let (expect_ok, res) = outcome.expect("task panicked");
+            let (ns, expect_ok, res) = outcome.expect("task panicked");
             if expect_ok {
-                assert!(res.is_ok(), "expected {:?} to be authorized", res);
+                assert!(
+                    res.is_ok(),
+                    "seed 7: namespace {ns} should be authorized: {res:?}"
+                );
             } else {
                 assert!(matches!(res, Err(UnauthorizedKubeconfigAccess())));
             }
diff --git a/src/resources.rs b/src/resources.rs
index 2e9bb91..256cd14 100644
--- a/src/resources.rs
+++ b/src/resources.rs
@@ -173,6 +173,98 @@ mod tests {
     use super::*;
     use rstest::rstest;
 
+    #[rstest]
+    #[case::uppercase("TRUE")]
+    #[case::mixed_case("True")]
+    #[case::numeric("1")]
+    #[case::whitespace(" true ")]
+    #[case::empty("")]
+    fn option_flags_require_exact_true(#[case] value: &str) {
+        let mut sync = crate::test_support::resource_sync();
+        sync.metadata.annotations = Some(std::collections::BTreeMap::from([
+            (FORCE_DELETE_ANNOTATION.into(), value.into()),
+            (DISABLE_TARGET_DELETION_ANNOTATION.into(), value.into()),
+        ]));
+        assert!(!sync.has_force_delete_option_enabled());
+        assert!(!sync.has_disable_target_deletion_option_enabled());
+    }
+
+    #[rstest]
+    #[case::core("v1", "", "v1")]
+    #[case::grouped("apps/v1", "apps", "v1")]
+    fn resource_reference_converts_type_metadata(
+        #[case] api_version: &str,
+        #[case] group: &str,
+        #[case] version: &str,
+    ) {
+        let reference = GVKN {
+            api_version: api_version.into(),
+            kind: "Deployment".into(),
+            name: "application".into(),
+        };
+        assert_eq!(
+            TypeMeta::from(&reference),
+            TypeMeta {
+                api_version: api_version.into(),
+                kind: "Deployment".into()
+            }
+        );
+        assert_eq!(
+            GroupVersionKind::try_from(&reference).expect("parse GVK"),
+            GroupVersionKind::gvk(group, version, "Deployment")
+        );
+    }
+
+    #[test]
+    fn resource_sync_serialization_uses_public_field_names_and_mapping_defaults() {
+        use serde_json::json;
+        let reference = json!({"resourceRef": {"apiVersion": "v1", "kind": "Secret", "name": "credentials"},
+            "cluster": {"namespace": "workloads", "kubeConfig": {"secretRef": {
+                "name": "remote-config", "namespace": "credentials", "key": "config"}}}});
+        let mut spec = json!({"source": reference, "target": reference});
+        let parsed: ResourceSyncSpec =
+            serde_json::from_value(spec.clone()).expect("deserialize omitted mappings");
+        assert!(parsed.mappings.is_empty());
+        assert_eq!(json!(parsed), spec);
+        spec["mappings"] = json!([]);
+        let parsed: ResourceSyncSpec =
+            serde_json::from_value(spec.clone()).expect("deserialize empty mappings");
+        assert!(parsed.mappings.is_empty());
+        assert!(json!(parsed).get("mappings").is_none());
+        spec["mappings"] = json!([{"fromFieldPath": "data", "toFieldPath": "spec"}]);
+        let parsed: ResourceSyncSpec =
+            serde_json::from_value(spec.clone()).expect("deserialize mappings");
+        assert_eq!(parsed.mappings[0].from_field_path.as_deref(), Some("data"));
+        assert_eq!(parsed.mappings[0].to_field_path.as_deref(), Some("spec"));
+        assert_eq!(json!(parsed), spec);
+    }
+
+    #[test]
+    fn container_schema_preserves_arbitrary_spec_in_every_version() {
+        let crd = SinkerContainer::crd_with_manual_schema();
+        assert_eq!(
+            crd.metadata.name.as_deref(),
+            Some("sinkercontainers.sinker.influxdata.io")
+        );
+        assert_eq!(crd.spec.scope, "Namespaced");
+        assert!(!crd.spec.versions.is_empty());
+        for version in crd.spec.versions {
+            let schema = version
+                .schema
+                .expect("version schema")
+                .open_api_v3_schema
+                .expect("OpenAPI schema");
+            assert_eq!(schema.type_.as_deref(), Some("object"));
+            assert_eq!(schema.required, Some(vec!["spec".to_string()]));
+            let properties = schema.properties.expect("schema properties");
+            let spec = properties.get("spec").expect("spec schema");
+            assert_eq!(spec.type_.as_deref(), Some("object"));
+            assert_eq!(spec.x_kubernetes_preserve_unknown_fields, Some(true));
+            assert!(spec.properties.is_none());
+            assert!(spec.additional_properties.is_none());
+        }
+    }
+
     macro_rules! gen_option_flag_tests {
         ($test_name:ident, $method:ident, $annotation_key:expr) => {
             #[rstest]
diff --git a/src/util.rs b/src/util.rs
index aa640e5..393cc52 100644
--- a/src/util.rs
+++ b/src/util.rs
@@ -35,3 +35,53 @@ impl<T> WithItemAdded<T> for Vec<T> {
         self
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use kube::runtime::controller::Action;
+    use rstest::rstest;
+    use std::time::Duration;
+
+    #[rstest]
+    #[case::empty(vec![], vec![])]
+    #[case::absent(vec!["first", "last"], vec!["first", "last"])]
+    #[case::duplicates(vec!["remove", "first", "remove", "last"], vec!["first", "last"])]
+    #[case::only_match(vec!["remove"], vec![])]
+    fn removing_items_preserves_other_items_and_order(
+        #[case] input: Vec<&str>,
+        #[case] expected: Vec<&str>,
+    ) {
+        assert_eq!(input.with_item_removed(&"remove"), expected);
+    }
+
+    #[test]
+    fn with_push_appends_without_deduplicating() {
+        assert_eq!(
+            Vec::new()
+                .with_push("first")
+                .with_push("second")
+                .with_push("first"),
+            vec!["first", "second", "first"]
+        );
+    }
+
+    #[test]
+    fn requeue_macro_supports_default_seconds_and_subsecond_durations() {
+        let default: crate::Result<Action> = requeue_after!();
+        let seconds: crate::Result<Action> = requeue_after!(in_seconds 12);
+        let duration: crate::Result<Action> = requeue_after!(Duration::from_millis(500));
+        assert_eq!(
+            default.expect("default requeue"),
+            Action::requeue(Duration::from_secs(5))
+        );
+        assert_eq!(
+            seconds.expect("seconds requeue"),
+            Action::requeue(Duration::from_secs(12))
+        );
+        assert_eq!(
+            duration.expect("duration requeue"),
+            Action::requeue(Duration::from_millis(500))
+        );
+    }
+}