use std::{sync::Arc, time::Duration};

use arc_swap::{ArcSwap, ArcSwapOption, Guard};
use tokio::{
    runtime::Handle,
    sync::Notify,
    time::{Instant, interval_at},
};
use kameleoon_trace_logging::trace_method;

use crate::{
    datafile::{datafile::DataFile, settings::Settings, streaming_service::StreamingService},
    debug_log,
    error::{ErrorCode, KErr},
    error_log,
    logging::{KameleoonLogger, LogLevel},
    network::{
        dto::DataFileDto,
        network_error::NetworkError,
        network_manager::{NetworkManager, datafile::HEADER_LAST_MODIFIED},
        response::Response,
        url_provider::UrlProvider,
    },
    types,
    utils::KRes,
};

// MARK: - DataFileManager

#[cfg_attr(test, mockall::automock(
    type DataFile=crate::datafile::datafile::MockDataFileImpl;
))]
pub(crate) trait DataFileManager: Send + Sync {
    type DataFile: DataFile;

    async fn initialize(self: &Arc<Self>) -> KRes<()>;
    fn datafile(&self) -> Guard<Arc<Self::DataFile>>;
    fn get_types_datafile(&self) -> Arc<types::DataFile>;
}

// MARK: - DataFileManagerImpl

type UpdateCb = Box<dyn Fn(&KRes<bool>) + Send + Sync>;

pub(crate) struct DataFileManagerImpl<D: DataFile + 'static, N: NetworkManager, S: StreamingService + 'static> {
    datafile: ArcSwap<D>,
    network_manager: Arc<N>,
    streaming_service: S,
    runtime_handle: Handle,
    stop_polling_notify: ArcSwapOption<Notify>,
    refresh_interval: Duration,
    environment: Option<String>,
    update_callback: UpdateCb,
    cached_types_datafile: ArcSwap<Option<Arc<types::DataFile>>>,
}

impl<D: DataFile, N: NetworkManager, S: StreamingService> DataFileManager for DataFileManagerImpl<D, N, S> {
    type DataFile = D;

    async fn initialize(self: &Arc<Self>) -> KRes<()> {
        //~ [42126]
        debug_log!("CALL DataFileManagerImpl::initialize()");
        //~ I'm not 100% sure the initialization loop handles init failures properly in both modes.
        self.select_update_mode(self.datafile.load().as_ref());
        let result = self.fetch_handle_store(None).await;
        debug_log!("RET DataFileManagerImpl::initialize() → {:?}", result);
        result.map(|_| ())
    }

    #[trace_method(LogLevel::Debug, skip(self, retval))]
    fn datafile(&self) -> Guard<Arc<Self::DataFile>> {
        self.datafile.load()
    }

    fn get_types_datafile(&self) -> Arc<types::DataFile> {
        // return cached
        if let Some(cached) = self.cached_types_datafile.load().as_ref().as_ref() {
            return cached.clone();
        }
        // compute from internal datafile
        let internal = self.datafile.load();
        let result = Arc::new(types::DataFile::from_internal(internal.as_ref()));
        self.cached_types_datafile.store(Arc::new(Some(result.clone())));

        result
    }
}

impl<D: DataFile, N: NetworkManager, S: StreamingService> Drop for DataFileManagerImpl<D, N, S> {
    #[trace_method(LogLevel::Debug, skip(self))]
    fn drop(&mut self) {
        self.stop_polling_mode();
        self.stop_streaming_mode();
    }
}

impl<D: DataFile, N: NetworkManager, S: StreamingService> DataFileManagerImpl<D, N, S> {
    pub fn new(
        network_manager: Arc<N>,
        streaming_service: S,
        datafile: Option<D>,
        refresh_interval: Duration,
        environment: Option<String>,
        update_callback: UpdateCb,
        runtime_handle: Handle,
    ) -> Self {
        Self {
            datafile: ArcSwap::from_pointee(datafile.unwrap_or_else(|| D::new_nullum(environment.clone()))),
            network_manager,
            streaming_service,
            runtime_handle,
            stop_polling_notify: ArcSwapOption::empty(),
            refresh_interval,
            environment,
            update_callback,
            cached_types_datafile: ArcSwap::new(Arc::new(None)),
        }
    }

    // MARK: - Fetch & Handle & Store

    async fn fetch_handle_store(self: &Arc<Self>, ts: Option<u64>) -> KRes<bool> {
        let result = match self.fetch(ts).await {
            Ok(response) => match response.status_code {
                304 => Ok(false),
                200 => self.handle(response).map(|datafile| self.store(datafile)).map(|_| true),
                _ => Err(KErr::new(
                    ErrorCode::Network,
                    format!(
                        "Unexpected status code receieved: {}, response body: {:?}",
                        response.status_code,
                        str::from_utf8(&response.body)
                    ),
                )),
            },
            Err(err) => Err(err.into()),
        };
        (self.update_callback)(&result);
        result
    }

    #[inline]
    async fn fetch(&self, ts: Option<u64>) -> Result<Response, NetworkError> {
        let environment = self.environment.as_deref();
        let datafile = self.datafile.load();
        let last_modified = datafile.as_ref().last_modified();

        self.network_manager
            .fetch_datafile(environment, ts, last_modified)
            .await
            .inspect_err(|err| error_log!("Fetch data file failed: {:?}", err))
    }

    #[inline]
    fn handle(self: &Arc<Self>, response: Response) -> KRes<D> {
        // Last Modified Header
        let last_modified = response.get_header_by_key(HEADER_LAST_MODIFIED).map(|s| s.to_owned());
        // Data File
        let datafile_dto =
            response.json::<DataFileDto>().inspect_err(|err| error_log!("Invalid data file response: {:?}", err))?;
        let datafile = D::new(datafile_dto, self.environment.clone(), last_modified);
        self.network_manager.url_provider().apply_url_domains(
            Some(datafile.settings().data_api_domain()),
            Some(datafile.settings().streaming_domain()),
            None,
        );
        self.select_update_mode(&datafile);
        Ok(datafile)
    }

    #[inline]
    fn store(&self, datafile: D) {
        self.datafile.store(Arc::new(datafile));
        self.cached_types_datafile.store(Arc::new(None));
    }

    // MARK: - Polling & Streaming mode

    fn select_update_mode(self: &Arc<Self>, datafile: &D) {
        if datafile.settings().is_streaming_mode() {
            self.stop_polling_mode();
            self.start_streaming_mode();
        } else {
            self.stop_streaming_mode();
            self.start_polling_mode();
        }
    }

    fn start_polling_mode(self: &Arc<Self>) {
        KameleoonLogger::info("Datafile polling mode: started");
        let notify = Arc::new(Notify::new());
        self.stop_polling_notify.store(Some(notify.clone()));
        let refresh_interval = self.refresh_interval;

        let weak = Arc::downgrade(self);
        self.runtime_handle.spawn(async move {
            let mut interval = interval_at(Instant::now() + refresh_interval, refresh_interval);

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        let Some(self_clone) = weak.upgrade() else {
                            break;
                        };
                        if let Err(err) = self_clone.fetch_handle_store(None).await {
                            error_log!("Polling fetch failed: {:?}", err);
                        }
                    }
                    _ = notify.notified() => {
                        break;
                    }
                }
            }
        });
    }

    fn stop_polling_mode(&self) {
        if let Some(notify) = self.stop_polling_notify.swap(None) {
            KameleoonLogger::info("Datafile polling mode: stopped");
            notify.notify_one();
        }
    }

    fn start_streaming_mode(self: &Arc<Self>) {
        if self.streaming_service.is_connected() {
            return;
        }
        KameleoonLogger::info("Datafile streaming mode: started");
        let weak = Arc::downgrade(self);
        self.streaming_service.connect(
            Box::new({
                let weak = weak.clone();
                let runtime_handle = self.runtime_handle.clone();
                move |ts| {
                    let weak = weak.clone();
                    runtime_handle.spawn(async move {
                        if let Some(this) = weak.upgrade() {
                            if let Err(err) = this.fetch_handle_store(Some(ts)).await {
                                error_log!("Streaming fetch failed: {:?}", err);
                            }
                        }
                    });
                }
            }),
            Box::new(move || {
                if let Some(this) = weak.upgrade() {
                    this.start_polling_mode();
                }
            }),
        );
    }

    fn stop_streaming_mode(&self) {
        if self.streaming_service.is_connected() {
            KameleoonLogger::info("Datafile streaming mode: stopped");
            self.streaming_service.disconnect();
        }
    }
}

// MARK: - Tests

#[cfg(test)]
mod tests {
    use rstest::rstest;
    use tokio::time::sleep;

    use crate::{
        datafile::{
            custom_data_info::MockCustomDataInfo,
            datafile_test_utils::{MockExperiment, MockFeatureFlag, MockMeGroup},
            segment::MockSegment,
            settings::MockSettings,
            variation_exposition::VariationByExposition,
        },
        network::{
            core_network_service::CoreNetworkServiceImpl,
            network_manager::NetworkManagerImpl,
            network_test_utils::{
                MockAccessTokenSource, MockNetProvider, MockUrlProvider, create_network_manager_net_provider,
                create_status_code_response, create_success_response_params,
            },
        },
        utils::test_utils::{TestDataFile, load_datafile, load_file},
    };

    use super::*;
    use std::{
        collections::HashMap,
        iter::repeat_with,
        ops::RangeInclusive,
        sync::atomic::{AtomicBool, AtomicI16, AtomicUsize, Ordering},
    };

    const DATA_FILE_OK: &str = "resources/datafiles/datafile_tndueuutdq.json";
    const DATA_FILE_FAIL: &str = "resources/datafiles/datafile_fail.json";
    const STREAMING_EVENT_TS: u64 = 1000;
    const ENV: &str = "env";

    // --- Mock DataFile ---
    #[derive(Debug)]
    struct MockDataFile {
        default: bool,
        last_modified: Option<String>,
        settings: MockSettings,
    }

    impl MockDataFile {
        fn new_mock(settings: MockSettings, last_modified: Option<&str>) -> Self {
            Self {
                default: false,
                settings,
                last_modified: last_modified.map(str::to_owned),
            }
        }

        fn make_default_settings() -> MockSettings {
            let mut mock_settings = MockSettings::new();
            mock_settings.expect_consent_required().returning(|| false);
            mock_settings.expect_is_streaming_mode().returning(|| false);
            mock_settings.expect_data_api_domain().return_const("data_api_domain".to_owned());
            mock_settings.expect_streaming_domain().return_const("streaming_domain".to_owned());
            return mock_settings;
        }
    }

    impl DataFile for MockDataFile {
        type FeatureFlag = MockFeatureFlag;
        type Experiment = MockExperiment;
        type Segment = MockSegment;
        type MeGroup = MockMeGroup;
        type Settings = MockSettings;
        type CustomDataInfo = MockCustomDataInfo;

        fn new(_dto: DataFileDto, _env: Option<String>, last_modified: Option<String>) -> Self {
            MockDataFile::new_mock(Self::make_default_settings(), last_modified.as_deref())
        }

        fn new_nullum(_env: Option<String>) -> Self {
            Self {
                default: true,
                ..Self::new_mock(Self::make_default_settings(), None)
            }
        }

        fn is_nullum(&self) -> bool {
            return self.default;
        }

        fn settings(&self) -> &Self::Settings {
            &self.settings
        }

        fn last_modified(&self) -> Option<&str> {
            self.last_modified.as_deref()
        }

        fn date_modified(&self) -> u64 {
            0
        }

        fn custom_data_info(&self) -> &Self::CustomDataInfo {
            todo!()
        }

        fn feature_flags(&self) -> &HashMap<String, Arc<Self::FeatureFlag>> {
            todo!()
        }

        fn feature_flags_by_id(&self) -> &HashMap<u32, Arc<Self::FeatureFlag>> {
            todo!()
        }

        fn ordered_feature_flags(&self) -> &[Arc<Self::FeatureFlag>] {
            todo!()
        }

        fn vars_by_id(&self) -> &HashMap<u32, Arc<dyn VariationByExposition>> {
            todo!()
        }

        fn segments(&self) -> &HashMap<u32, Arc<Self::Segment>> {
            todo!()
        }

        fn audience_tracking_segments(&self) -> &[Arc<Self::Segment>] {
            todo!()
        }

        fn me_groups(&self) -> &HashMap<String, Self::MeGroup> {
            todo!()
        }

        fn holdout(&self) -> Option<&Self::Experiment> {
            todo!()
        }

        fn has_targeted_delivery_rule(&self) -> bool {
            todo!()
        }

        fn get_feature_flag(&self, _feature_key: &str) -> KRes<&Arc<Self::FeatureFlag>> {
            todo!()
        }

        fn has_experiment_js_css_variable(&self, _experiment_id: u32) -> bool {
            todo!()
        }

        fn get_info_by_exp_id(&self, _experiment_id: u32) -> Option<&crate::datafile::datafile::InfoByExpId> {
            todo!()
        }
    }

    // --- MockStreamingService ---
    #[derive(Debug, Default)]
    struct MockStreamingService {
        connected: Arc<AtomicBool>,
    }

    impl StreamingService for MockStreamingService {
        fn connect(&self, on_msg: Box<dyn Fn(u64) + Sync + Send>, _on_disconnect: Box<dyn FnOnce() + Send + Sync>) {
            self.connected.store(true, Ordering::Relaxed);
            on_msg(STREAMING_EVENT_TS);
        }
        fn disconnect(&self) {
            self.connected.store(false, Ordering::Relaxed);
        }
        fn is_connected(&self) -> bool {
            self.connected.load(Ordering::Relaxed)
        }
    }

    // --- Helpers ---
    type TestDataFileManager = DataFileManagerImpl<
        MockDataFile,
        NetworkManagerImpl<CoreNetworkServiceImpl<MockNetProvider>, MockUrlProvider, MockAccessTokenSource>,
        MockStreamingService,
    >;

    type TypesDataFileManager = DataFileManagerImpl<
        TestDataFile,
        NetworkManagerImpl<CoreNetworkServiceImpl<MockNetProvider>, MockUrlProvider, MockAccessTokenSource>,
        MockStreamingService,
    >;

    fn make_datafile_manager_params(
        datafile: Option<MockDataFile>,
        mock_url_provider: MockUrlProvider,
        response_file_name: &str,
        refresh_interval: Option<Duration>,
        number_response: Option<usize>,
        responses: Option<Vec<Response>>,
        update_callback: UpdateCb,
        runtime_handle: Handle,
    ) -> Arc<TestDataFileManager> {
        let response_json = load_file(response_file_name);
        let responses: Option<Vec<_>> =
            responses.map(|r| r.into_iter().map(|rr| Ok::<Response, NetworkError>(rr)).collect());
        let responses: Vec<_> = responses.unwrap_or(
            repeat_with(|| Ok(create_success_response_params(&response_json, vec![])))
                .take(number_response.unwrap_or(1))
                .collect(),
        );
        let net_provider = MockNetProvider::with_responses(responses);

        let network_manager =
            create_network_manager_net_provider(Some(net_provider.clone()), Some(mock_url_provider), None);

        Arc::new(DataFileManagerImpl::new(
            Arc::new(network_manager),
            MockStreamingService::default(),
            datafile,
            refresh_interval.unwrap_or(Duration::from_millis(1000)),
            Some(ENV.into()),
            update_callback,
            runtime_handle,
        ))
    }

    fn make_datafile_manager() -> Arc<
        DataFileManagerImpl<
            MockDataFile,
            NetworkManagerImpl<CoreNetworkServiceImpl<MockNetProvider>, MockUrlProvider, MockAccessTokenSource>,
            MockStreamingService,
        >,
    > {
        let (url_provider, _, _) = make_mock_url_provider(0..=1, 0..=1, None);
        let cb: UpdateCb = Box::new(|_| {});
        make_datafile_manager_params(None, url_provider, DATA_FILE_OK, None, None, None, cb, Handle::current())
    }

    fn make_types_datafile_manager(datafile: TestDataFile, runtime_handle: Handle) -> Arc<TypesDataFileManager> {
        let (url_provider, _, _) = make_mock_url_provider(0..=0, 0..=0, None);
        let network_manager = create_network_manager_net_provider(
            Some(MockNetProvider::with_responses(vec![])),
            Some(url_provider),
            None,
        );

        Arc::new(DataFileManagerImpl::new(
            Arc::new(network_manager),
            MockStreamingService::default(),
            Some(datafile),
            Duration::from_millis(1000),
            Some(ENV.into()),
            Box::new(|_| {}),
            runtime_handle,
        ))
    }

    fn make_mock_url_provider(
        datafile_url_times: RangeInclusive<usize>,
        apply_url_domains_times: RangeInclusive<usize>,
        last_modified_param: Option<u64>,
    ) -> (MockUrlProvider, Arc<AtomicUsize>, Arc<AtomicUsize>) {
        let make_datafile_url_counter = Arc::new(AtomicUsize::new(0));
        let make_datafile_url_counter_closure = make_datafile_url_counter.clone();
        let apply_url_domains_counter = Arc::new(AtomicUsize::new(0));
        let apply_url_domains_counter_closure = apply_url_domains_counter.clone();
        let mut mock_url_provider: MockUrlProvider = MockUrlProvider::new();
        mock_url_provider
            .expect_make_datafile_url()
            .withf(move |env_arg, last_modified| *env_arg == Some(ENV) && *last_modified == last_modified_param)
            .returning(move |_, _| {
                make_datafile_url_counter_closure.fetch_add(1, Ordering::Relaxed);
                "https://api.example.com/".to_string()
            })
            .times(datafile_url_times);
        mock_url_provider
            .expect_apply_url_domains()
            .withf(|data_api_domain, streaming_domain, _| {
                *data_api_domain == Some("data_api_domain") && *streaming_domain == Some("streaming_domain")
            })
            .returning(move |_, _, _| {
                apply_url_domains_counter_closure.fetch_add(1, Ordering::Relaxed);
                ()
            })
            .times(apply_url_domains_times);
        (mock_url_provider, make_datafile_url_counter, apply_url_domains_counter)
    }

    // --- Tests ---

    #[rstest]
    #[case::ok(DATA_FILE_OK, 1, true)]
    #[case::fail(DATA_FILE_FAIL, 0, false)]
    #[tokio::test]
    async fn test_initialize(#[case] df: &str, #[case] apply_url_domains: usize, #[case] expect_success: bool) {
        // Arrange
        let (url_provider, _, _) = make_mock_url_provider(1..=1, apply_url_domains..=apply_url_domains, None);
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let cb: UpdateCb = Box::new(move |ur| {
            let _ = tx.send(ur.is_ok());
        });
        let mgr = make_datafile_manager_params(None, url_provider, df, None, None, None, cb, Handle::current());
        // Act
        let res = mgr.initialize().await;
        // Assert
        assert_eq!(expect_success, res.is_ok());
        assert_eq!(!expect_success, mgr.datafile().is_nullum());
        rx.close();
        assert_eq!(expect_success, rx.try_recv().unwrap());
        assert!(rx.is_empty());
    }

    #[tokio::test]
    async fn test_initialize_already_has_datafile() {
        let (url_provider, _, _) = make_mock_url_provider(1..=1, 1..=1, None);
        let mut settings = MockSettings::default();
        settings.expect_is_streaming_mode().returning(|| false);
        let datafile = Some(MockDataFile::new_mock(settings, None));
        let cb: UpdateCb = Box::new(|_| {});
        let mgr =
            make_datafile_manager_params(datafile, url_provider, DATA_FILE_OK, None, None, None, cb, Handle::current());
        let res = mgr.initialize().await;
        assert!(res.is_ok());
        assert!(!mgr.datafile().is_nullum());
    }

    #[tokio::test]
    async fn test_handle_last_modified() {
        // Arrange
        let mgr = make_datafile_manager();
        let mut headers = Vec::new();
        let last_modified = "Tue, 26 Aug 2025 11:41:45 GMT";
        headers.push((HEADER_LAST_MODIFIED.to_string(), last_modified.to_string()));
        let resp = create_success_response_params(&load_file(DATA_FILE_OK), headers);

        // Act
        let df = mgr.handle(resp).unwrap();

        // Assert
        assert_eq!(last_modified, df.last_modified().unwrap());
    }

    #[test]
    fn test_store() {
        // Arrange
        let (url_provider, _, _) = make_mock_url_provider(0..=0, 0..=0, None);
        let cb: UpdateCb = Box::new(|_| {});
        let runtime = tokio::runtime::Runtime::new().unwrap();
        let mgr = make_datafile_manager_params(
            None,
            url_provider,
            DATA_FILE_OK,
            None,
            None,
            None,
            cb,
            runtime.handle().clone(),
        );
        let mock_datafile = MockDataFile::new_mock(MockSettings::default(), Some("ts"));

        // Act
        mgr.store(mock_datafile);

        // Assert
        assert_eq!(Some("ts"), mgr.datafile().as_ref().last_modified());
    }

    #[test]
    fn test_get_types_datafile_cache_value_and_recreate_after_store() {
        let runtime = tokio::runtime::Runtime::new().unwrap();
        let mgr = make_types_datafile_manager(load_datafile(DATA_FILE_OK), runtime.handle().clone());

        let initial = mgr.get_types_datafile();
        assert!(mgr.cached_types_datafile.load().as_ref().is_some());

        mgr.store(<TestDataFile as DataFile>::new_nullum(Some(ENV.into())));

        assert!(mgr.cached_types_datafile.load().as_ref().is_none());

        let recreated = mgr.get_types_datafile();
        let cached = mgr.cached_types_datafile.load().as_ref().as_ref().unwrap().clone();

        assert!(recreated.feature_flags.is_empty());
        assert!(!Arc::ptr_eq(&initial, &recreated));
        assert!(Arc::ptr_eq(&recreated, &cached));
    }

    #[tokio::test]
    async fn test_select_update_mode_switches() {
        let (url_provider, _, _) = make_mock_url_provider(0..=0, 0..=0, None);
        let cb: UpdateCb = Box::new(|_| {});
        let mgr =
            make_datafile_manager_params(None, url_provider, DATA_FILE_OK, None, None, None, cb, Handle::current());

        fn run_case(mgr: &Arc<TestDataFileManager>, is_streaming: bool) {
            let mut mock_settings = MockSettings::new();
            mock_settings.expect_is_streaming_mode().return_const(is_streaming);
            let df = MockDataFile::new_mock(mock_settings, None);

            mgr.select_update_mode(&df);

            if is_streaming {
                assert!(mgr.streaming_service.is_connected());
                assert!(mgr.stop_polling_notify.load().is_none());
            } else {
                assert!(!mgr.streaming_service.is_connected());
                assert!(mgr.stop_polling_notify.load().is_some());
            }
        }

        // Run in sequence: Streaming → Polling → Streaming
        run_case(&mgr, true);
        run_case(&mgr, false);
        run_case(&mgr, true);
    }

    #[tokio::test]
    #[rstest(case(true), case(false))]
    async fn test_cleanup_stops_modes(#[case] is_streaming: bool) {
        // Arrange
        const INTERVAL: Duration = Duration::from_millis(250);
        let (url_provider, _, _) = make_mock_url_provider(1..=1, 1..=1, None);
        let update_count = Arc::new(AtomicI16::new(0));
        let update_count_clone = update_count.clone();
        let cb: UpdateCb = Box::new(move |_| {
            update_count_clone.fetch_add(1, Ordering::Relaxed);
        });
        let mgr = make_datafile_manager_params(
            None,
            url_provider,
            DATA_FILE_OK,
            Some(INTERVAL),
            None,
            None,
            cb,
            Handle::current(),
        );
        let is_streaming_service_connected = mgr.streaming_service.connected.clone();
        let _ = mgr.initialize().await;

        let mut mock_settings = MockSettings::new();
        mock_settings.expect_is_streaming_mode().return_const(is_streaming);
        let df = MockDataFile::new_mock(mock_settings, None);

        mgr.select_update_mode(&df);
        // Act-Assert
        if is_streaming {
            assert!(mgr.streaming_service.is_connected());
            assert!(is_streaming_service_connected.load(Ordering::Relaxed));
        } else {
            assert!(mgr.stop_polling_notify.load().is_some());
            assert!(!is_streaming_service_connected.load(Ordering::Relaxed));
        }
        drop(mgr);
        assert!(!is_streaming_service_connected.load(Ordering::Relaxed));
        assert_eq!(1, update_count.load(Ordering::Relaxed));
        sleep(INTERVAL + INTERVAL / 2).await;
        assert_eq!(1, update_count.load(Ordering::Relaxed));
    }

    #[tokio::test]
    async fn test_streaming_mode_update_by_event() {
        // Arrange
        let (url_provider, _, _) = make_mock_url_provider(1..=1, 1..=1, Some(STREAMING_EVENT_TS));
        let cb: UpdateCb = Box::new(|_| {});
        let mgr =
            make_datafile_manager_params(None, url_provider, DATA_FILE_OK, None, None, None, cb, Handle::current());

        let mut mock_settings = MockSettings::new();
        mock_settings.expect_is_streaming_mode().return_const(true);
        let df = MockDataFile::new_mock(mock_settings, None);
        mgr.select_update_mode(&df);

        sleep(Duration::from_millis(1)).await;
    }

    #[tokio::test]
    async fn test_polling_mode_update_by_refresh_interval() {
        let (min_count, max_count) = (5, 21);
        // We can't rely on mock times(), because checking of number of calls happens on Drop for mock struct,
        // but in the case if Drop failed -> the test method returns success anyway.
        let (url_provider, make_datafile_counter, apply_domains_counter) =
            make_mock_url_provider(0..=100, 0..=100, None);

        let refresh_interval = Some(Duration::from_millis(2));
        let cb: UpdateCb = Box::new(|_| {});
        let mgr = make_datafile_manager_params(
            None,
            url_provider,
            DATA_FILE_OK,
            refresh_interval,
            Some(max_count),
            None,
            cb,
            Handle::current(),
        );
        let _ = mgr.initialize().await;
        sleep(Duration::from_millis(15)).await;

        let make_datafile_counter = make_datafile_counter.load(Ordering::Relaxed);
        let apply_domains_counter = apply_domains_counter.load(Ordering::Relaxed);
        assert!(
            min_count <= make_datafile_counter && make_datafile_counter <= max_count,
            "make_datafile_counter: {}",
            make_datafile_counter,
        );
        assert!(
            min_count <= apply_domains_counter && apply_domains_counter <= max_count,
            "apply_domains_counter: {}",
            apply_domains_counter,
        );
    }

    #[rstest]
    #[case::status_500(create_status_code_response(500), false, true, 1..=1, false)]
    #[case::status_304(create_status_code_response(304), true, true, 1..=1, false)]
    #[case::status_200(create_success_response_params(&load_file(DATA_FILE_OK), vec![]), true, true, 2..=2, true)]
    #[tokio::test]
    async fn test_status_code_responses(
        #[case] second_response: Response,
        #[case] expect_ok: bool,
        #[case] expect_datafile: bool,
        #[case] apply_domain_range: RangeInclusive<usize>,
        #[case] expect_update_callback: bool,
    ) {
        // Arrange
        let response_json = load_file(DATA_FILE_OK);
        let (url_provider, _, _) = make_mock_url_provider(2..=2, apply_domain_range, None);
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let cb: UpdateCb = Box::new(move |ur| {
            let _ = tx.send(ur.clone());
        });
        let mgr = make_datafile_manager_params(
            None,
            url_provider,
            DATA_FILE_OK,
            None,
            None,
            Some(vec![create_success_response_params(&response_json, vec![]), second_response]),
            cb,
            Handle::current(),
        );

        let _ = mgr.fetch_handle_store(None).await;
        let res = mgr.fetch_handle_store(None).await;

        // Assert
        rx.close();
        assert!(rx.try_recv().is_ok());
        if expect_ok {
            assert!(res.is_ok());
            let upd_value = rx.try_recv().ok().unwrap().is_ok_and(|b| b);
            assert_eq!(expect_update_callback, upd_value);
        } else {
            assert!(res.is_err());
            assert!(!rx.try_recv().is_err());
        }
        assert!(rx.is_empty());

        assert_eq!(expect_datafile, !mgr.datafile().is_nullum());
    }
}
