package cmd

import (
	"context"
	"errors"
	"math"
	"path/filepath"
	"testing"
	"time"

	"github.com/bruin-data/bruin/pkg/date"
	"github.com/bruin-data/bruin/pkg/logger"
	"github.com/bruin-data/bruin/pkg/pipeline"
	"github.com/bruin-data/bruin/pkg/scheduler"
	"github.com/spf13/afero"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest"
)

func TestClean(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name string
		s    string
		want string
	}{
		{
			name: "empty",
			s:    "\u001B[0m\u001B[94m[2023-12-05 19:11:25] [hello_python] >> [2023-12-05 19:11:25 - INFO] Starting extractor: gcs_bucket_files",
			want: "[2023-12-05 19:11:25] [hello_python] >> [2023-12-05 19:11:25 - INFO] Starting extractor: gcs_bucket_files",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()
			assert.Equalf(t, tt.want, Clean(tt.s), "Clean(%v)", tt.s)
		})
	}
}

func BenchmarkClean(b *testing.B) {
	for range make([]struct{}, b.N) {
		// Clean("\u001B[0m\u001B[94m[2023-12-05 19:11:25] [hello_python] >> [2023-12-05 19:11:25 - INFO] Starting extractor: gcs_bucket_files")
		Clean("\u001B[0m\u001B[94m[2023-12-05 19:11:26] [hello_python] >>   File \"/Users/burak/.bruin/virtualenvs/77ef9663d804ac96afe6fb2a10d2b2b817a07fd82875759af8910b4fe31a7149/lib/python3.9/site-packages/google/api_core/page_iterator.py\", line 208, in _items_iter")
	}
}

var (
	task0 = &pipeline.Asset{
		Name: "Task0",
		Type: pipeline.AssetTypeBigqueryQuery,
		Tags: []string{"tag1"},
		Columns: []pipeline.Column{
			{
				Name:        "Column0",
				Type:        "STRING",
				Description: "A sample column",
				PrimaryKey:  true,
				Checks: []pipeline.ColumnCheck{
					{Name: "Check0"},
				},
			},
		},
		CustomChecks: []pipeline.CustomCheck{
			{Name: "CustomCheck0", Description: "some custom check"},
		},
	}

	task1 = &pipeline.Asset{
		Name: "Task1",
		Type: pipeline.AssetTypePython,
		Tags: []string{"tag1", "x"},
	}

	task2 = &pipeline.Asset{
		Name: "Task2",
		Type: pipeline.AssetTypeBigqueryQuery,
		Tags: []string{"tag2"},
	}

	task3 = &pipeline.Asset{
		Name: "Task3", Type: pipeline.AssetTypePython,
		Tags:      []string{"tag4"},
		Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task9"}},
	}

	task4 = &pipeline.Asset{
		Name:      "Task4",
		Type:      pipeline.AssetTypeBigqueryQuery,
		Tags:      []string{"tag1", "tag3"},
		Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task9"}},
	}

	task5 = &pipeline.Asset{
		Name:      "Task5",
		Type:      pipeline.AssetTypeBigqueryQuery,
		Tags:      []string{"tag1"},
		Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task8"}}, // Task8 is upstream
	}

	task6 = &pipeline.Asset{
		Name:      "Task6",
		Type:      pipeline.AssetTypeBigqueryQuery,
		Tags:      []string{},
		Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task8"}}, // Task8 is upstream
	}

	task7 = &pipeline.Asset{
		Name:      "Task7",
		Type:      pipeline.AssetTypePython,
		Tags:      []string{""},
		Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task8"}}, // Task8 is upstream
	}

	task8 = &pipeline.Asset{
		Name:      "Task8",
		Type:      pipeline.AssetTypeBigqueryQuery,
		Tags:      []string{"tag1", "exclude"},
		Upstreams: []pipeline.Upstream{}, // No upstreams for Task8
	}

	task9 = &pipeline.Asset{
		Name:      "Task9",
		Type:      pipeline.AssetTypeBigqueryQuery,
		Tags:      []string{"tag4"},
		Upstreams: []pipeline.Upstream{},
	}

	task10 = &pipeline.Asset{
		Name: "Task10",
		Type: pipeline.AssetTypeBigqueryQuery,
		Tags: []string{"tag1"},
		Columns: []pipeline.Column{
			{
				Name:        "Column10",
				Type:        "STRING",
				Description: "A sample column",
				PrimaryKey:  true,
				Checks: []pipeline.ColumnCheck{
					{Name: "Check10"},
				},
			},
		},
		CustomChecks: []pipeline.CustomCheck{
			{Name: "CustomCheck10", Description: "some custom check"},
		},
	}
	task11 = &pipeline.Asset{
		Name: "Task11",
		Type: pipeline.AssetTypeBigqueryQuery,
		Tags: []string{"tag1", "tag4"},
		Columns: []pipeline.Column{
			{
				Name:        "Column11",
				Type:        "STRING",
				Description: "A sample column",
				PrimaryKey:  true,
				Checks: []pipeline.ColumnCheck{
					{Name: "Check11"},
				},
			},
		},
		CustomChecks: []pipeline.CustomCheck{
			{Name: "CustomCheck11", Description: "some custom check"},
		},
		Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task9"}},
	}
	task12 = &pipeline.Asset{
		Name: "Task12",
		Type: pipeline.AssetTypeBigqueryQuery,
		Tags: []string{},
		Columns: []pipeline.Column{
			{
				Name:        "Column12",
				Type:        "STRING",
				Description: "A sample column",
				PrimaryKey:  true,
				Checks: []pipeline.ColumnCheck{
					{Name: "Check12"},
				},
			},
		},
		CustomChecks: []pipeline.CustomCheck{
			{Name: "CustomCheck12", Description: "some custom check"},
		},
		Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task9"}},
	}
)

func TestApplyFilters(t *testing.T) { //nolint
	t.Parallel() // Enable parallel execution for the top-level test
	tests := []struct {
		name            string
		pipeline        *pipeline.Pipeline
		filter          *Filter
		expectedPending []string
		expectError     bool
		expectedError   string
	}{
		{
			name: "Single Task",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{Name: "Task1", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag1"}},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{IncludeDownstream: false},
			expectedPending: []string{"Task1"},
			expectError:     false,
		},
		{
			name: "Include Tag Matches",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{Name: "Task1", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag1", "tag2"}},
					{Name: "Task2", Type: pipeline.AssetTypePython, Tags: []string{"tag2"}},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{IncludeTag: "tag1"},
			expectedPending: []string{"Task1"},
			expectError:     false,
		},
		{
			name: "Include Tag No Matches",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{Name: "Task1", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag1", "tag2"}},
					{Name: "Task2", Type: pipeline.AssetTypePython, Tags: []string{"tag2"}},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{IncludeTag: "tag3"},
			expectedPending: nil,
			expectError:     true,
			expectedError:   "no assets found with include tag 'tag3'",
		},
		{
			name: "Include Tag And Downstream",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{
						Name:      "Task1",
						Type:      pipeline.AssetTypePython,
						Tags:      []string{"tag1"},
						Upstreams: []pipeline.Upstream{},
					},
					{
						Name:      "Task2",
						Type:      pipeline.AssetTypeBigqueryQuery,
						Tags:      []string{},
						Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task1"}},
					},
					{
						Name:      "Task3",
						Type:      pipeline.AssetTypePython,
						Tags:      []string{},
						Upstreams: []pipeline.Upstream{{Type: "asset", Value: "Task2"}},
					},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{IncludeDownstream: true, IncludeTag: "tag1"},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot use the --downstream flag when running the whole pipeline",
		},
		{
			name: "Only Check Task Type",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{Name: "MainTask", Type: pipeline.AssetTypePython},
					{Name: "CheckTask", Type: pipeline.AssetTypeBigqueryQuery},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{OnlyTaskTypes: []string{"checks"}},
			expectedPending: []string{},
			expectError:     false,
		},
		{
			name: "Invalid Task Type",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{Name: "MainTask", Type: pipeline.AssetTypePython},
					{Name: "CheckTask", Type: pipeline.AssetTypeBigqueryQuery},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:        &Filter{OnlyTaskTypes: []string{"bigquery-query"}},
			expectedError: "invalid value for '--only' flag: 'bigquery-query', available values are 'main', 'checks', and 'push-metadata'",
			expectError:   true,
		},
		{
			name: "Empty Pipeline",
			pipeline: &pipeline.Pipeline{
				Name:   "EmptyPipeline",
				Assets: []*pipeline.Asset{},
			},
			filter:          &Filter{},
			expectedPending: []string{},
			expectError:     false,
		},
		{
			name: "Empty Filter - Regular workflow",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{Name: "Task1", Type: pipeline.AssetTypePython, Tags: []string{"tag1"}},
					{Name: "Task2", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag2"}},
					{Name: "Task3", Type: pipeline.AssetTypePython, Tags: []string{}},
					{Name: "Task4", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag1", "tag3"}},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{},
			expectedPending: []string{"Task1", "Task2", "Task3", "Task4"},
			expectError:     false,
		},
		{
			name: "Include Tag And Only Checks",
			pipeline: &pipeline.Pipeline{
				Name: "ComplexPipeline",
				Assets: []*pipeline.Asset{
					{Name: "Task1", Type: pipeline.AssetTypePython, Tags: []string{"common-tag"}},
					{Name: "Task2", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"common-tag"}},
					{Name: "Task3", Type: pipeline.AssetTypePython, Tags: []string{"unique-tag"}},
					{Name: "Task4", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"common-tag"}},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{IncludeTag: "common-tag", OnlyTaskTypes: []string{"checks"}},
			expectedPending: []string{},
			expectError:     false,
		},
		{
			name: "Multiple Tags - Mixed Matches",
			pipeline: &pipeline.Pipeline{
				Name: "MultiTagPipeline",
				Assets: []*pipeline.Asset{
					{Name: "Task1", Type: pipeline.AssetTypePython, Tags: []string{"tag1", "tag2"}},
					{Name: "Task2", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag2", "tag3"}},
					{Name: "Task3", Type: pipeline.AssetTypePython, Tags: []string{"tag3", "tag4"}},
					{Name: "Task4", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag1", "tag4"}},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{IncludeTag: "tag1"},
			expectedPending: []string{"Task1", "Task4"}, // Only tasks with "tag1" should match
			expectError:     false,
		},
		{
			name: "Empty Tags",
			pipeline: &pipeline.Pipeline{
				Name: "EmptyTagsPipeline",
				Assets: []*pipeline.Asset{
					{Name: "Task1", Type: pipeline.AssetTypePython, Tags: []string{}},
					{Name: "Task2", Type: pipeline.AssetTypeBigqueryQuery, Tags: []string{"tag2"}},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter:          &Filter{IncludeTag: "tag1"},
			expectedPending: []string{}, // No matches since "tag1" isn't present
			expectError:     true,
			expectedError:   "no assets found with include tag 'tag1'",
		},
		{
			name: " Run single Task in a full pipeline",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				SingleTask: task4,
			},
			expectedPending: []string{"Task4"},
			expectError:     false,
		},
		{
			name: "Single Task with a Tag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				SingleTask: task4,
				IncludeTag: "tag1",
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "you cannot use the '--tag' flag when running a single asset",
		},

		{
			name: "Run single Task with downstream",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				SingleTask:        task8,
				IncludeDownstream: true,
			},
			expectedPending: []string{"Task8", "Task7", "Task6", "Task5"},
			expectError:     false,
		},
		{
			name: "Run single Task with downstream and a tag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				SingleTask:        task8,
				IncludeDownstream: true,
				IncludeTag:        "tag1",
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "you cannot use the '--tag' flag when running a single asset",
		},
		{
			name: "Filter By Tag and Only at the same time",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: true},
			},
			filter: &Filter{
				IncludeDownstream: false,
				IncludeTag:        "tag1",
				OnlyTaskTypes:     []string{"push-metadata"},
				PushMetaData:      true,
			},
			expectedPending: []string{"Task8:metadata-push", "Task5:metadata-push", "Task4:metadata-push"},
			expectError:     false,
		},
		{
			name: "Filter By Tag and run its downstream",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				IncludeDownstream: true,
				IncludeTag:        "tag1",
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot use the --downstream flag when running the whole pipeline",
		},
		{
			name: "Filter by Tag and Run Matching Downstream Tasks by Type",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: true},
			},
			filter: &Filter{
				IncludeDownstream: true,
				IncludeTag:        "tag4",
				PushMetaData:      true,
				OnlyTaskTypes:     []string{"push-metadata"},
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot use the --downstream flag when running the whole pipeline",
		},

		{
			name: "Run single asset with PushmetaData",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: true},
			},
			filter: &Filter{
				SingleTask:        task4,
				IncludeDownstream: true,
				PushMetaData:      true,
			},
			expectedPending: []string{"Task4", "Task4:metadata-push"},
			expectError:     false,
		},

		{
			name: " Run single asset with PushmetaData and Only flag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: true},
			},
			filter: &Filter{
				SingleTask:        task4,
				IncludeDownstream: true,
				PushMetaData:      true,
				OnlyTaskTypes:     []string{"push-metadata"},
			},
			expectedPending: []string{"Task4:metadata-push"},
			expectError:     false,
		},
		{
			name: " Run only Checks",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
					task10,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: true},
			},
			filter: &Filter{
				IncludeDownstream: false,
				PushMetaData:      true,
				OnlyTaskTypes:     []string{"checks"},
			},
			expectedPending: []string{"Task0:Column0:Check0", "Task0:custom-check:customcheck0", "Task10:Column10:Check10", "Task10:custom-check:customcheck10"},
			expectError:     false,
		},

		{
			name: " Run only Main",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
					task10,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: true},
			},
			filter: &Filter{
				IncludeDownstream: false,
				PushMetaData:      true,
				OnlyTaskTypes:     []string{"main"}, // checks and metadata should be excluded
			},
			expectedPending: []string{"Task0", "Task1", "Task2", "Task3", "Task4", "Task5", "Task6", "Task7", "Task8", "Task9", "Task10"},
			expectError:     false,
		},

		{
			name: " Run full pipeline",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
					task10,
				},
			},
			filter: &Filter{},
			expectedPending: []string{
				"Task0", "Task1", "Task2", "Task3", "Task4", "Task5", "Task6", "Task7", "Task8", "Task9", "Task10",
				"Task10:Column10:Check10", "Task10:custom-check:customcheck10", "Task0:Column0:Check0", "Task0:custom-check:customcheck0",
			},
			expectError: false,
		},
		{
			name: " Run single asset with only Checks",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task6,
					task7,
					task8,
					task9,
					task10,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: true},
			},
			filter: &Filter{
				SingleTask:        task10,
				IncludeDownstream: true,
				PushMetaData:      true,
				OnlyTaskTypes:     []string{"checks"},
			},
			expectedPending: []string{"Task10:Column10:Check10", "Task10:custom-check:customcheck10"},
			expectError:     false,
		},
		{
			name: "Only flag with multiple arguments",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task6,
					task7,
					task8,
					task9,
					task10,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: true},
			},
			filter: &Filter{
				SingleTask:        task10,
				IncludeDownstream: true,
				PushMetaData:      true,
				OnlyTaskTypes:     []string{"checks", "push-metadata"},
			},
			expectedPending: []string{"Task10:Column10:Check10", "Task10:custom-check:customcheck10", "Task10:metadata-push"},
			expectError:     false,
		},

		{
			name: "Single Task with a exclude-tag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				SingleTask: task4,
				ExcludeTag: "tag1",
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "when running a single asset with '--exclude-tag', you must also use the '--downstream' flag",
		},

		{
			name: "Full pipeline with exclude-tag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag: "tag1",
			},
			expectedPending: []string{"Task2", "Task3"},
			expectError:     false,
		},

		{
			name: "Full pipeline with exclude-tag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag: "non-existent-tag",
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "no assets found with exclude tag 'non-existent-tag'",
		},
		{
			name: "Exclude Tag and the downstreams",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:        "exclude",
				IncludeDownstream: true,
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot use the --downstream flag when running the whole pipeline",
		},
		{
			name: "Exclude Tag and the downstreams",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:        "exclude",
				IncludeDownstream: true,
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot use the --downstream flag when running the whole pipeline",
		},

		{
			name: "Exclude and Include Tag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:        "tag3",
				IncludeTag:        "tag1",
				IncludeDownstream: false,
			},
			expectedPending: []string{"Task5", "Task8"},
		},

		{
			name: "Exclude and Include Tag (exclude tag non existent)",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:        "non-existent-tag",
				IncludeTag:        "tag1",
				IncludeDownstream: false,
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "no assets found with exclude tag 'non-existent-tag'",
		},

		{
			name: "Exclude ,Include Tag with downstream",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task2,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:        "tag3",
				IncludeTag:        "tag1",
				IncludeDownstream: true,
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot use the --downstream flag when running the whole pipeline",
		},

		{
			name: "Exclude with only checks ",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:    "x",
				OnlyTaskTypes: []string{"checks"},
			},
			expectedPending: []string{"Task0:Column0:Check0", "Task0:custom-check:customcheck0"},
		},
		{
			name: "Exclude with only checks 2",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:    "x",
				OnlyTaskTypes: []string{"checks"},
				PushMetaData:  true,
			},
			expectedPending: []string{"Task0:Column0:Check0", "Task0:custom-check:customcheck0"},
		},

		{
			name: "Exclude with only checks and metadapush",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
				},
				MetadataPush: pipeline.MetadataPush{Global: true, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:    "x",
				OnlyTaskTypes: []string{"checks", "push-metadata"},
				PushMetaData:  true,
			},
			expectedPending: []string{"Task0:Column0:Check0", "Task0:custom-check:customcheck0", "Task0:metadata-push"},
		},

		{
			name: "Exclude ,Include Tag with downstream and only flag",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task0,
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
					task9,
					task10,
					task11,
					task12,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:        "tag1",
				IncludeTag:        "tag4",
				IncludeDownstream: true,
				OnlyTaskTypes:     []string{"checks"},
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot use the --downstream flag when running the whole pipeline",
		},
		{
			name: "Exclude Tag with single asset and the downstream",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					task1,
					task2,
					task3,
					task4,
					task5,
					task6,
					task7,
					task8,
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				ExcludeTag:        "exclude",
				IncludeDownstream: true,
				SingleTask:        task8,
			},
			expectedPending: []string{"Task7", "Task6", "Task5"},
		},
		{
			name: "Skip all tasks except single column check",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{
						Name: "Task1",
						Type: pipeline.AssetTypeBigqueryQuery,
						Columns: []pipeline.Column{
							{
								Name: "Column1",
								Checks: []pipeline.ColumnCheck{
									{ID: "check-123", Name: "Check1"},
								},
							},
						},
					},
					{
						Name: "Task2",
						Type: pipeline.AssetTypePython,
					},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				singleCheckID: "check-123",
			},
			expectedPending: []string{"Task1:Column1:Check1"},
			expectError:     false,
		},
		{
			name: "Skip all tasks except single custom check",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{
						Name: "Task1",
						Type: pipeline.AssetTypeBigqueryQuery,
						CustomChecks: []pipeline.CustomCheck{
							{ID: "custom-123", Name: "CustomCheck1"},
						},
					},
					{
						Name: "Task2",
						Type: pipeline.AssetTypePython,
					},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				singleCheckID: "custom-123",
			},
			expectedPending: []string{"Task1:custom-check:customcheck1"},
			expectError:     false,
		},
		{
			name: "Skip all tasks with non-existent check ID",
			pipeline: &pipeline.Pipeline{
				Name: "TestPipeline",
				Assets: []*pipeline.Asset{
					{
						Name: "Task1",
						Type: pipeline.AssetTypeBigqueryQuery,
						CustomChecks: []pipeline.CustomCheck{
							{ID: "custom-123", Name: "CustomCheck1"},
						},
					},
				},
				MetadataPush: pipeline.MetadataPush{Global: false, BigQuery: false},
			},
			filter: &Filter{
				singleCheckID: "non-existent-id",
			},
			expectedPending: []string{},
			expectError:     true,
			expectedError:   "cannot find check with the given ID",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel() // Enable parallel execution for individual test cases
			logger := zap.NewNop().Sugar()
			s := scheduler.NewScheduler(logger, tt.pipeline, "test")
			err := ApplyAllFilters(context.Background(), tt.filter, s, tt.pipeline)

			// Assert
			if tt.expectError {
				require.Error(t, err)
				assert.EqualError(t, err, tt.expectedError)
			} else {
				require.NoError(t, err)
				markedTasks := s.GetTaskInstancesByStatus(scheduler.Pending)
				assert.Len(t, markedTasks, len(tt.expectedPending))

				// Collect task HumanIDs
				taskNames := []string{}
				for _, task := range markedTasks {
					taskNames = append(taskNames, task.GetHumanID()) // Use HumanID for accurate matching
					t.Log(task.GetHumanID())
				}

				// Check if each expected task is present in the marked tasks
				for _, name := range tt.expectedPending {
					assert.Contains(t, taskNames, name)
				}
			}
		})
	}
}

func TestParseDate(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name          string
		startDateStr  string
		endDateStr    string
		expectError   bool
		expectedStart time.Time
		expectedEnd   time.Time
	}{
		{
			name:          "Valid Dates",
			startDateStr:  "2023-12-01",
			endDateStr:    "2023-12-31",
			expectError:   false,
			expectedStart: time.Date(2023, 12, 1, 0, 0, 0, 0, time.UTC),
			expectedEnd:   time.Date(2023, 12, 31, 0, 0, 0, 0, time.UTC),
		},
		{
			name:         "Invalid Start Date",
			startDateStr: "invalid-date",
			endDateStr:   "2023-12-31",
			expectError:  true,
		},
		{
			name:         "Invalid End Date",
			startDateStr: "2023-12-01",
			endDateStr:   "invalid-date",
			expectError:  true,
		},
		{
			name:          "Valid Date with Time",
			startDateStr:  "2023-12-01 15:04:05",
			endDateStr:    "2023-12-31 23:59:59",
			expectError:   false,
			expectedStart: time.Date(2023, 12, 1, 15, 4, 5, 0, time.UTC),
			expectedEnd:   time.Date(2023, 12, 31, 23, 59, 59, 0, time.UTC),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()
			logger := zaptest.NewLogger(t).Sugar()
			startDate, endDate, err := ParseDate(tt.startDateStr, tt.endDateStr, logger)

			if tt.expectError {
				assert.Error(t, err)
			} else {
				require.NoError(t, err)
				assert.Equal(t, tt.expectedStart, startDate)
				assert.Equal(t, tt.expectedEnd, endDate)
			}
		})
	}
}

func TestValidation(t *testing.T) {
	t.Parallel()
	logger := zaptest.NewLogger(t).Sugar()

	tests := []struct {
		name              string
		runConfig         *scheduler.RunConfig
		inputPath         string
		expectedInputPath string
		expectError       bool
		expectedError     string
	}{
		{
			name: "Invalid Start Date",
			runConfig: &scheduler.RunConfig{
				StartDate: "invalid-date",
				EndDate:   "2023-12-31",
			},
			inputPath:   "some/path",
			expectError: true,
		},
		{
			name: "Invalid End Date",
			runConfig: &scheduler.RunConfig{
				StartDate: "2023-12-01",
				EndDate:   "invalid-date",
			},
			inputPath:   "some/path",
			expectError: true,
		},
		{
			name: "Valid Input",
			runConfig: &scheduler.RunConfig{
				StartDate: "2023-12-01",
				EndDate:   "2023-12-31",
			},
			inputPath:         "some/path",
			expectedInputPath: "some/path",
			expectError:       false,
		},
		{
			name: "valid input without a path, should default to the current path",
			runConfig: &scheduler.RunConfig{
				StartDate: "2023-12-01",
				EndDate:   "2023-12-31",
			},
			inputPath:         "",
			expectedInputPath: ".",
			expectError:       false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel() // Enable parallel execution for individual test cases
			startDate, endDate, path, err := ValidateRunConfig(tt.runConfig, tt.inputPath, logger)

			if tt.expectError {
				require.Error(t, err)
				if tt.expectedError != "" {
					require.EqualError(t, err, tt.expectedError)
				}
			} else {
				require.NoError(t, err)
				assert.Equal(t, time.Date(2023, 12, 1, 0, 0, 0, 0, time.UTC), startDate)
				assert.Equal(t, time.Date(2023, 12, 31, 0, 0, 0, 0, time.UTC), endDate)
				assert.Equal(t, tt.expectedInputPath, path)
			}
		})
	}
}

func TestCheckLintFunc(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name          string
		foundPipeline *pipeline.Pipeline
		pipelinePath  string
	}{
		{
			name: "Lint Rule Error",
			foundPipeline: &pipeline.Pipeline{
				Name:        "TestPipeline",
				Concurrency: 1,
			},
			pipelinePath: "path/to/pipeline",
		},
		{
			name: "Linting Error",
			foundPipeline: &pipeline.Pipeline{
				Name:        "TestPipeline",
				Concurrency: 1,
			},
			pipelinePath: "path/to/pipeline",
		},
	}

	for _, tt := range tests {
		ctx := context.Background()
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()
			logger := zaptest.NewLogger(t).Sugar()

			err := CheckLint(ctx, tt.foundPipeline, tt.pipelinePath, logger, false)
			require.NoError(t, err, "Expected no error but got one")
		})
	}
}

func TestReadState(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name              string
		statePath         string
		stateContent      string
		expectedError     bool
		filter            *Filter
		expectedFilter    *Filter
		runConfig         *scheduler.RunConfig
		expectedRunConfig *scheduler.RunConfig
	}{
		{
			name:      "Valid State",
			statePath: "/logs/run",
			stateContent: `{
				"parameters": {
					"tag": "test-tag",
					"only": ["task-type"],
					"pushMetadata": true,
					"excludeTag": "exclude-tag"
				}
			}`,
			expectedError: false,
			filter:        &Filter{},
			expectedFilter: &Filter{
				IncludeTag:    "test-tag",
				OnlyTaskTypes: []string{"task-type"},
				PushMetaData:  true,
				ExcludeTag:    "exclude-tag",
			},
			runConfig:         &scheduler.RunConfig{},
			expectedRunConfig: &scheduler.RunConfig{},
		},
		{
			name:              "Empty State",
			statePath:         "/logs/run",
			stateContent:      `{}`,
			expectedError:     false,
			filter:            &Filter{},
			expectedFilter:    &Filter{},
			runConfig:         &scheduler.RunConfig{},
			expectedRunConfig: &scheduler.RunConfig{},
		},
		{
			name:      "Pre-filled Filter Values",
			statePath: "/logs/run",
			stateContent: `{
				"parameters": {
					"tag": "new-tag",
					"only": ["new-task-type"],
					"pushMetadata": false,
					"excludeTag": "new-exclude-tag"
				}
			}`,
			expectedError: false,
			filter: &Filter{
				IncludeTag:    "old-tag",
				OnlyTaskTypes: []string{"old-task-type"},
				PushMetaData:  true,
				ExcludeTag:    "old-exclude-tag",
			},
			expectedFilter: &Filter{
				IncludeTag:    "new-tag",
				OnlyTaskTypes: []string{"new-task-type"},
				PushMetaData:  false,
				ExcludeTag:    "new-exclude-tag",
			},
			runConfig:         &scheduler.RunConfig{},
			expectedRunConfig: &scheduler.RunConfig{},
		},
		{
			name:      "Partial Pre-filled Filter Values",
			statePath: "/logs/run",
			stateContent: `{
				"parameters": {
					"tag": "partial-tag"
				}
			}`,
			expectedError: false,
			filter: &Filter{
				IncludeTag:    "existing-tag",
				OnlyTaskTypes: []string{"existing-task-type"},
				PushMetaData:  true,
				ExcludeTag:    "existing-exclude-tag",
			},
			expectedFilter: &Filter{
				IncludeTag:   "partial-tag",
				PushMetaData: false,
				ExcludeTag:   "",
			},
			runConfig:         &scheduler.RunConfig{},
			expectedRunConfig: &scheduler.RunConfig{},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()
			fs := afero.NewMemMapFs()
			err := afero.WriteFile(fs, filepath.Join(tt.statePath, "2024_12_19_22_59_13.json"), []byte(tt.stateContent), 0o644)
			require.NoError(t, err)

			filter := tt.filter
			runConfig := tt.runConfig

			pipelineState, err := ReadState(fs, tt.statePath, filter)

			if tt.expectedError {
				require.Error(t, err)
				assert.Nil(t, pipelineState)
			} else {
				require.NoError(t, err)
				assert.NotNil(t, pipelineState)
				assert.Equal(t, tt.expectedFilter, filter)
				assert.Equal(t, tt.expectedRunConfig, runConfig)
			}
		})
	}
}

type mockAssetCounter struct {
	count int
}

func (m *mockAssetCounter) GetAssetCountWithTasksPending() int {
	return m.count
}

func TestValidate_ShouldValidateFalse_ReturnsNil(t *testing.T) {
	t.Parallel()
	err := Validate(false, &mockAssetCounter{count: 1}, nil, context.Background(), &pipeline.Pipeline{}, "path", zap.NewNop().Sugar())
	if err != nil {
		t.Errorf("expected nil, got %v", err)
	}
}

func TestValidate_AssetsPending_CallsLintCheckerWithFalse(t *testing.T) {
	t.Parallel()
	called := false
	var validateOnlyAssetLevel bool

	lintChecker := func(ctx context.Context, foundPipeline *pipeline.Pipeline, pipelinePath string, logger logger.Logger, voal bool) error {
		called = true
		validateOnlyAssetLevel = voal
		return nil
	}

	err := Validate(true, &mockAssetCounter{count: 2}, lintChecker, context.Background(), &pipeline.Pipeline{}, "path", zap.NewNop().Sugar())
	if err != nil {
		t.Errorf("expected nil, got %v", err)
	}
	if !called {
		t.Error("expected lintChecker to be called")
	}
	if validateOnlyAssetLevel {
		t.Error("expected validateOnlyAssetLevel to be false when assets are pending")
	}
}

func TestValidate_NoAssetsPending_CallsLintCheckerWithTrue(t *testing.T) {
	t.Parallel()
	called := false
	var validateOnlyAssetLevel bool

	lintChecker := func(ctx context.Context, foundPipeline *pipeline.Pipeline, pipelinePath string, logger logger.Logger, voal bool) error {
		called = true
		validateOnlyAssetLevel = voal
		return nil
	}

	err := Validate(true, &mockAssetCounter{count: 0}, lintChecker, context.Background(), &pipeline.Pipeline{}, "path", zap.NewNop().Sugar())
	if err != nil {
		t.Errorf("expected nil, got %v", err)
	}
	if !called {
		t.Error("expected lintChecker to be called")
	}
	if !validateOnlyAssetLevel {
		t.Error("expected validateOnlyAssetLevel to be true when no assets are pending")
	}
}

func TestValidate_LintCheckerReturnsError_PropagatesError(t *testing.T) {
	t.Parallel()
	expectedErr := errors.New("lint error")
	lintChecker := func(ctx context.Context, foundPipeline *pipeline.Pipeline, pipelinePath string, logger logger.Logger, voal bool) error {
		return expectedErr
	}

	err := Validate(true, &mockAssetCounter{count: 0}, lintChecker, context.Background(), &pipeline.Pipeline{}, "path", zap.NewNop().Sugar())
	if !errors.Is(err, expectedErr) {
		t.Errorf("expected %v, got %v", expectedErr, err)
	}
}

func TestFullRefreshWithStartDateFlags(t *testing.T) {
	t.Parallel()

	tests := []struct {
		fullRefresh       bool
		name              string
		pipelineStartDate string
		flagStartDate     string
		flagEndDate       string
		expectedStartDate time.Time
		expectedEndDate   time.Time
		expectedError     bool
	}{
		// 1. Start with non-full refresh tests (simpler cases)
		{
			fullRefresh:       false,
			name:              "non-full refresh: empty pipeline, no flags -> use yesterday",
			pipelineStartDate: "",
			flagStartDate:     "",
			flagEndDate:       "",
			expectedStartDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, time.UTC)
			}(),
			expectedEndDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999999999, time.UTC)
			}(),
		},
		{
			fullRefresh:       false,
			name:              "non-full refresh: empty pipeline, no flags -> use default (yesterday)",
			pipelineStartDate: "2023-06-15",
			flagStartDate:     "",
			flagEndDate:       "",
			expectedStartDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, time.UTC)
			}(),
			expectedEndDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999999999, time.UTC)
			}(),
		},
		{
			fullRefresh:       false,
			name:              "non-full refresh: pipeline start date, use flag start date",
			pipelineStartDate: "2023-06-15",
			flagStartDate:     "2023-01-15",
			flagEndDate:       "",
			expectedStartDate: time.Date(2023, 1, 15, 0, 0, 0, 0, time.UTC),
			expectedEndDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999999999, time.UTC)
			}(),
		},
		{
			fullRefresh:       false,
			name:              "non-full refresh: end flag only -> yesterday start, flag end",
			pipelineStartDate: "2023-06-15",
			flagStartDate:     "",
			flagEndDate:       "2024-05-20",
			expectedStartDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, time.UTC)
			}(),
			expectedEndDate: time.Date(2024, 5, 20, 0, 0, 0, 0, time.UTC),
			expectedError:   true, // end date is older than start date
		},
		{
			fullRefresh:       false,
			name:              "non-full refresh: empty pipeline, both flags -> use flags",
			pipelineStartDate: "",
			flagStartDate:     "2024-02-01",
			flagEndDate:       "2024-02-28",
			expectedStartDate: time.Date(2024, 2, 1, 0, 0, 0, 0, time.UTC),
			expectedEndDate:   time.Date(2024, 2, 28, 0, 0, 0, 0, time.UTC),
		},
		{
			fullRefresh:       false,
			name:              "non-full refresh: both flags -> override everything",
			pipelineStartDate: "2023-06-15",
			flagStartDate:     "2024-03-01",
			flagEndDate:       "2024-03-31",
			expectedStartDate: time.Date(2024, 3, 1, 0, 0, 0, 0, time.UTC),
			expectedEndDate:   time.Date(2024, 3, 31, 0, 0, 0, 0, time.UTC),
		},
		// 2. Gradually introduce full refresh tests - minimal data first
		{
			fullRefresh:       true,
			name:              "full refresh: empty pipeline, no flags -> use yesterday",
			pipelineStartDate: "",
			flagStartDate:     "",
			flagEndDate:       "",
			expectedStartDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, time.UTC)
			}(),
			expectedEndDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999999999, time.UTC)
			}(),
		},
		{
			fullRefresh:       true,
			name:              "full refresh: pipeline only, no flags -> use pipeline start , end date is yesterday",
			pipelineStartDate: "2023-06-15",
			flagStartDate:     "",
			flagEndDate:       "",
			expectedStartDate: time.Date(2023, 6, 15, 0, 0, 0, 0, time.UTC),
			expectedEndDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999999999, time.UTC)
			}(),
		},
		// 3. Mix complexity - single flags
		{
			fullRefresh:       true,
			name:              "full refresh: start flag only -> use pipeline start, yesterday end",
			pipelineStartDate: "2023-06-15",
			flagStartDate:     "2024-02-01",
			flagEndDate:       "",
			expectedStartDate: time.Date(2023, 0o6, 15, 0, 0, 0, 0, time.UTC),
			expectedEndDate: func() time.Time {
				yesterday := time.Now().AddDate(0, 0, -1)
				return time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999999999, time.UTC)
			}(),
		},
		{
			fullRefresh:       true,
			name:              "full refresh: end flag only -> use pipeline start, flag end",
			pipelineStartDate: "2023-06-15",
			flagStartDate:     "",
			flagEndDate:       "2024-02-28",
			expectedStartDate: time.Date(2023, 6, 15, 0, 0, 0, 0, time.UTC),
			expectedEndDate:   time.Date(2024, 2, 28, 0, 0, 0, 0, time.UTC),
		},
		// 4. More complex - both flags but empty pipeline
		{
			fullRefresh:       true,
			name:              "full refresh: empty pipeline, both flags -> use flags",
			pipelineStartDate: "",
			flagStartDate:     "2024-04-01",
			flagEndDate:       "2024-04-30",
			expectedStartDate: time.Date(2024, 4, 1, 0, 0, 0, 0, time.UTC),
			expectedEndDate:   time.Date(2024, 4, 30, 0, 0, 0, 0, time.UTC),
		},
		// 5. Most complex - full data with edge cases
		{
			fullRefresh:       true,
			name:              "full refresh: both flags + pipeline -> pipeline start, flag end",
			pipelineStartDate: "2023-01-01",
			flagStartDate:     "2024-01-01",
			flagEndDate:       "2024-01-31",
			expectedStartDate: time.Date(2023, 1, 1, 0, 0, 0, 0, time.UTC),
			expectedEndDate:   time.Date(2024, 1, 31, 0, 0, 0, 0, time.UTC),
		},
		{
			fullRefresh:       true,
			name:              "full refresh: pipeline start overrides flag dates",
			pipelineStartDate: "2024-05-01",
			flagStartDate:     "2024-03-01",
			flagEndDate:       "2024-03-31",
			expectedStartDate: time.Date(2024, 5, 1, 0, 0, 0, 0, time.UTC),
			expectedEndDate:   time.Date(2024, 3, 31, 0, 0, 0, 0, time.UTC),
			expectedError:     true, // end date is older than start date
		},
		{
			fullRefresh:       true,
			name:              "full refresh: pipeline start always wins over flags",
			pipelineStartDate: "2024-01-01",
			flagStartDate:     "2024-05-01",
			flagEndDate:       "2024-05-31",
			expectedStartDate: time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
			expectedEndDate:   time.Date(2024, 5, 31, 0, 0, 0, 0, time.UTC),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			// Create a test pipeline with start_date
			testPipeline := &pipeline.Pipeline{
				Name:        "TestPipeline",
				StartDate:   tt.pipelineStartDate,
				Assets:      []*pipeline.Asset{},
				Concurrency: 1,
			}

			// Test the actual implementation
			logger := zaptest.NewLogger(t).Sugar()

			// Set default flag values if not provided
			cliStartDate := tt.flagStartDate
			cliEndDate := tt.flagEndDate
			if cliStartDate == "" {
				yesterday := time.Now().AddDate(0, 0, -1)
				cliStartDate = time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, time.UTC).Format("2006-01-02 15:04:05.000000")
			}
			if cliEndDate == "" {
				yesterday := time.Now().AddDate(0, 0, -1)
				cliEndDate = time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999999999, time.UTC).Format("2006-01-02 15:04:05.999999")
			}

			// Call the actual implementation
			actualStartDate, err := DetermineStartDate(cliStartDate, testPipeline, tt.fullRefresh, logger)

			// Parse end date separately
			actualEndDate, endDateErr := date.ParseTime(cliEndDate)

			if tt.expectedError {
				// For error cases, we might get dates but they should be logically invalid
				// (e.g., end date before start date)
				if err == nil && endDateErr == nil {
					// If no parsing error, check if end date is before start date
					assert.True(t, actualEndDate.Before(actualStartDate) || actualEndDate.Equal(actualStartDate), "Expected logical error: end date should be before or equal to start date")
				}
				return
			}

			require.NoError(t, err, "Should not error for valid test case")
			require.NoError(t, endDateErr, "Should not error for valid end date")

			// Compare dates - allow some tolerance for time precision
			assert.True(t, actualStartDate.Equal(tt.expectedStartDate) ||
				math.Abs(float64(actualStartDate.Sub(tt.expectedStartDate))) < float64(time.Second),
				"Start date mismatch: expected %v, got %v", tt.expectedStartDate, actualStartDate)

			assert.True(t, actualEndDate.Equal(tt.expectedEndDate) ||
				math.Abs(float64(actualEndDate.Sub(tt.expectedEndDate))) < float64(time.Second),
				"End date mismatch: expected %v, got %v", tt.expectedEndDate, actualEndDate)

			// Pipeline's default start_date should still be available
			assert.Equal(t, tt.pipelineStartDate, testPipeline.StartDate, "Pipeline should retain its default start_date")
		})
	}
}

func TestApplyIntervalModifiersWithFullRefresh(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name                   string
		fullRefresh            bool
		applyIntervalModifiers bool
		expectedApplyModifiers bool
		expectedWarning        bool
	}{
		{
			name:                   "full-refresh true, apply-interval-modifiers true -> should ignore apply-modifiers and show warning",
			fullRefresh:            true,
			applyIntervalModifiers: true,
			expectedApplyModifiers: false,
			expectedWarning:        true,
		},
		{
			name:                   "full-refresh false, apply-interval-modifiers true -> should keep apply-modifiers",
			fullRefresh:            false,
			applyIntervalModifiers: true,
			expectedApplyModifiers: true,
			expectedWarning:        false,
		},
		{
			name:                   "full-refresh true, apply-interval-modifiers false -> no warning",
			fullRefresh:            true,
			applyIntervalModifiers: false,
			expectedApplyModifiers: false,
			expectedWarning:        false,
		},
		{
			name:                   "full-refresh false, apply-interval-modifiers false -> no warning",
			fullRefresh:            false,
			applyIntervalModifiers: false,
			expectedApplyModifiers: false,
			expectedWarning:        false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			// Simulate the logic from run.go
			fullRefresh := tt.fullRefresh
			applyIntervalModifiers := tt.applyIntervalModifiers

			var warningShown bool

			if fullRefresh && applyIntervalModifiers {
				// In the real code, this would print to warningPrinter
				// For testing, we just set a flag
				warningShown = true
				applyIntervalModifiers = false
			}

			// Test expectations
			assert.Equal(t, tt.expectedApplyModifiers, applyIntervalModifiers,
				"ApplyIntervalModifiers should be %v but was %v", tt.expectedApplyModifiers, applyIntervalModifiers)
			assert.Equal(t, tt.expectedWarning, warningShown,
				"Warning should be shown: %v but was %v", tt.expectedWarning, warningShown)
		})
	}
}
