package watchsync import ( "context" "encoding/base64" "errors" "strings" "testing" "time" pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1" "github.com/Silo-Server/silo-server/internal/historyimport" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" ) const ( testPluginProviderKey = "plugin:4:anilist" testPluginCapabilityID = "anilist" testWatchHistoryID = "history-1" testSecondHistoryID = "history-2" testPlaybackSessionID = "playback-1" testEpisodeMediaID = "episode-1" ) type fakeWatchSyncPluginClient struct { exchangeResponse *pluginv1.WatchSyncCredentialResponse refreshResponse *pluginv1.WatchSyncCredentialResponse accountResponse *pluginv1.WatchSyncGetAccountResponse applyResponse *pluginv1.WatchSyncApplyEventsResponse deviceStartResponse *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse devicePollResponse *pluginv1.WatchSyncDeviceAuthorizationServicePollResponse listResponse *pluginv1.WatchSyncListRemoteStateResponse listResponses []*pluginv1.WatchSyncListRemoteStateResponse applyErr error applyRequest *pluginv1.WatchSyncApplyEventsRequest refreshRequest *pluginv1.WatchSyncRefreshCredentialsRequest accountRequest *pluginv1.WatchSyncGetAccountRequest deviceStartRequest *pluginv1.WatchSyncDeviceAuthorizationServiceStartRequest devicePollRequest *pluginv1.WatchSyncDeviceAuthorizationServicePollRequest listRequests []*pluginv1.WatchSyncListRemoteStateRequest } func (f *fakeWatchSyncPluginClient) StartDeviceAuthorization(_ context.Context, req *pluginv1.WatchSyncDeviceAuthorizationServiceStartRequest) (*pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse, error) { f.deviceStartRequest = req if f.deviceStartResponse != nil { return f.deviceStartResponse, nil } return &pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse{}, nil } func (f *fakeWatchSyncPluginClient) PollDeviceAuthorization(_ context.Context, req *pluginv1.WatchSyncDeviceAuthorizationServicePollRequest) (*pluginv1.WatchSyncDeviceAuthorizationServicePollResponse, error) { f.devicePollRequest = req if f.devicePollResponse != nil { return f.devicePollResponse, nil } return &pluginv1.WatchSyncDeviceAuthorizationServicePollResponse{}, nil } func (f *fakeWatchSyncPluginClient) ExchangeAPIKey(_ context.Context, _ *pluginv1.WatchSyncExchangeAPIKeyRequest) (*pluginv1.WatchSyncCredentialResponse, error) { return f.exchangeResponse, nil } func (f *fakeWatchSyncPluginClient) RefreshCredentials(_ context.Context, req *pluginv1.WatchSyncRefreshCredentialsRequest) (*pluginv1.WatchSyncCredentialResponse, error) { f.refreshRequest = req if f.refreshResponse != nil { return f.refreshResponse, nil } return &pluginv1.WatchSyncCredentialResponse{}, nil } func (f *fakeWatchSyncPluginClient) GetAccount(_ context.Context, req *pluginv1.WatchSyncGetAccountRequest) (*pluginv1.WatchSyncGetAccountResponse, error) { f.accountRequest = req if f.accountResponse != nil { return f.accountResponse, nil } return &pluginv1.WatchSyncGetAccountResponse{Account: &pluginv1.WatchSyncAccount{ExternalSubject: testProviderAccountID}}, nil } func (f *fakeWatchSyncPluginClient) ApplyEvents(_ context.Context, req *pluginv1.WatchSyncApplyEventsRequest) (*pluginv1.WatchSyncApplyEventsResponse, error) { f.applyRequest = req return f.applyResponse, f.applyErr } func (f *fakeWatchSyncPluginClient) ListRemoteState(_ context.Context, req *pluginv1.WatchSyncListRemoteStateRequest) (*pluginv1.WatchSyncListRemoteStateResponse, error) { f.listRequests = append(f.listRequests, req) if len(f.listResponses) > 0 { response := f.listResponses[0] f.listResponses = f.listResponses[1:] return response, nil } if f.listResponse != nil { return f.listResponse, nil } return &pluginv1.WatchSyncListRemoteStateResponse{}, nil } type fakePluginCredentialRepository struct { saved Connection err error } func (r *fakePluginCredentialRepository) UpsertConnection(_ context.Context, conn Connection) (Connection, error) { if r.err != nil { return Connection{}, r.err } r.saved = conn return conn, nil } func testPluginProvider(t *testing.T, client WatchSyncPluginClient) *PluginProvider { t.Helper() return testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ExportWatched: true, MaxBatchSize: 25, }) } func testPluginProviderWithDescriptor(t *testing.T, client WatchSyncPluginClient, descriptor *pluginv1.WatchSyncProviderDescriptor) *PluginProvider { t.Helper() provider, err := NewPluginProvider(PluginProviderOptions{ InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID, DisplayName: "AniList", Descriptor: descriptor, ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) { return client, nil }, }) if err != nil { t.Fatal(err) } return provider } func TestPluginProviderUsesConnectionSpecificHistorySource(t *testing.T) { provider := testPluginProvider(t, &fakeWatchSyncPluginClient{}) if got := provider.HistorySource(); got != testPluginProviderKey { t.Fatalf("HistorySource() = %q, want %q", got, testPluginProviderKey) } } func TestPluginProviderRejectsUnsupportedInitialDescriptor(t *testing.T) { _, err := NewPluginProvider(PluginProviderOptions{ InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID, Descriptor: &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_AUTHORIZATION_CODE}, ExportWatched: true, }, ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) { return nil, nil }, }) if err == nil { t.Fatal("expected authorization-code-only descriptor to be rejected") } _, err = NewPluginProvider(PluginProviderOptions{ InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID, Descriptor: &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ExportWatched: true, SupportedMediaTypes: []pluginv1.WatchSyncMediaType{pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_UNSPECIFIED}, }, ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) { return nil, nil }, }) if err == nil { t.Fatal("expected unsupported media descriptor to be rejected") } _, err = NewPluginProvider(PluginProviderOptions{ InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID, Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{ pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_DEVICE_CODE, pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY, }}, ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) { return nil, nil }, }) if err == nil || !strings.Contains(err.Error(), "multiple host authentication methods") { t.Fatalf("multiple auth methods error = %v", err) } } func TestPluginProviderConnectsAPIKeyWithoutPersistingInPluginConfig(t *testing.T) { client := &fakeWatchSyncPluginClient{exchangeResponse: &pluginv1.WatchSyncCredentialResponse{ Credentials: &pluginv1.WatchSyncCredentials{AccessToken: testValidatedToken, TokenType: testBearerTokenType}, Account: &pluginv1.WatchSyncAccount{ExternalSubject: "7", Username: testPluginUsername}, }} provider := testPluginProvider(t, client) if provider.Key() != testPluginProviderKey { t.Fatalf("provider key = %q", provider.Key()) } tokens, account, err := provider.ConnectWithAPIKey(context.Background(), "input-token") if err != nil { t.Fatal(err) } if tokens.AccessToken != testValidatedToken || account.ID != "7" || account.Username != testPluginUsername { t.Fatalf("tokens=%#v account=%#v", tokens, account) } } func TestPluginProviderRejectsMissingAccountIdentity(t *testing.T) { for _, subject := range []string{"", " \t\n "} { client := &fakeWatchSyncPluginClient{exchangeResponse: &pluginv1.WatchSyncCredentialResponse{ Credentials: &pluginv1.WatchSyncCredentials{AccessToken: testValidatedToken, TokenType: testBearerTokenType}, Account: &pluginv1.WatchSyncAccount{ExternalSubject: subject, Username: testPluginUsername}, }} provider := testPluginProvider(t, client) if _, _, err := provider.ConnectWithAPIKey(context.Background(), "input-token"); err == nil || !strings.Contains(err.Error(), "account identity") { t.Fatalf("subject %q: error = %v", subject, err) } } } func TestPluginProviderRefreshReturnsCredentialsAlongsideFault(t *testing.T) { client := &fakeWatchSyncPluginClient{refreshResponse: &pluginv1.WatchSyncCredentialResponse{ Credentials: &pluginv1.WatchSyncCredentials{AccessToken: testRotatedAccessToken, TokenType: testBearerTokenType}, Fault: &pluginv1.WatchSyncFault{ Code: pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_INVALID_CREDENTIAL, SafeMessage: "credential rotated-access rejected; reconnect required", }, }} provider := testPluginProvider(t, client) tokens, err := provider.RefreshToken(context.Background(), ServerConfig{}, Connection{ AccessToken: testOldAccessToken, RefreshToken: testOldRefreshToken, }) if tokens.AccessToken != testRotatedAccessToken || tokens.RefreshToken != "" || tokens.TokenExpiresAt != nil { t.Fatalf("tokens = %#v", tokens) } if !isWatchSyncInvalidCredentialError(err) { t.Fatalf("error = %#v", err) } if strings.Contains(err.Error(), testRotatedAccessToken) || !strings.Contains(err.Error(), "[REDACTED]") { t.Fatalf("returned credentials were not redacted: %q", err) } } func TestPluginProviderExportsRichEpisodeIdentity(t *testing.T) { client := &fakeWatchSyncPluginClient{} client.applyResponse = &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: testWatchHistoryID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED, }}} provider := testPluginProvider(t, client) result, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{AccessToken: testSecretValue}, []LocalPlay{{ HistoryID: testWatchHistoryID, MediaItemID: testEpisodeMediaID, Kind: historyimport.KindEpisode, SeriesTVDBID: "123", SeriesTMDBID: "456", SeasonNumber: 2, EpisodeNumber: 7, WatchedAt: time.Now().UTC(), DurationSeconds: 1440, }}) if err != nil { t.Fatal(err) } if len(result.Sent) != 1 || result.Sent[0] != testWatchHistoryID { t.Fatalf("result = %#v", result) } event := client.applyRequest.GetEvents()[0] if client.applyRequest.GetContext().GetCredentials().GetAccessToken() != testSecretValue || event.GetMedia().GetSeriesExternalIds()["tvdb"] != "123" || event.GetMedia().GetEpisodeNumber() != 7 || event.GetMedia().GetMediaType() != pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_EPISODE { t.Fatalf("apply request = %#v", client.applyRequest) } } func TestPluginProviderBatchesEventsInOneRPC(t *testing.T) { client := &fakeWatchSyncPluginClient{applyResponse: &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{ {EventId: testWatchHistoryID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED}, {EventId: testSecondHistoryID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_REJECTED}, }}} provider := testPluginProvider(t, client) result, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{ {HistoryID: testWatchHistoryID, Kind: historyimport.KindMovie}, {HistoryID: testSecondHistoryID, Kind: historyimport.KindMovie}, }) if err != nil { t.Fatal(err) } if len(client.applyRequest.GetEvents()) != 2 || len(result.Sent) != 1 || len(result.NotFound) != 1 { t.Fatalf("request=%#v result=%#v", client.applyRequest, result) } } func TestPluginProviderRejectsUnspecifiedExportMedia(t *testing.T) { client := &fakeWatchSyncPluginClient{} provider := testPluginProvider(t, client) result, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{{HistoryID: testWatchHistoryID}}) if err != nil { t.Fatal(err) } if client.applyRequest != nil { t.Fatalf("unexpected apply request = %#v", client.applyRequest) } if result.Failed[testWatchHistoryID] != watchSyncUnsupportedMediaMessage { t.Fatalf("result = %#v", result) } } func TestPluginProviderSkipsUnsupportedExportMedia(t *testing.T) { client := &fakeWatchSyncPluginClient{applyResponse: &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{ {EventId: testWatchHistoryID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED}, }}} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ExportWatched: true, MaxBatchSize: 25, SupportedMediaTypes: []pluginv1.WatchSyncMediaType{pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE}, }) result, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{ {HistoryID: testWatchHistoryID, Kind: historyimport.KindMovie}, {HistoryID: testSecondHistoryID, Kind: historyimport.KindEpisode}, }) if err != nil { t.Fatal(err) } if len(client.applyRequest.GetEvents()) != 1 || client.applyRequest.GetEvents()[0].GetEventId() != testWatchHistoryID { t.Fatalf("apply request = %#v", client.applyRequest) } if got := result.Failed[testSecondHistoryID]; got != watchSyncUnsupportedEpisodeMediaMessage { t.Fatalf("result = %#v", result) } } func TestPluginProviderMapsPerEventRateLimitAndKeepsSuccesses(t *testing.T) { client := &fakeWatchSyncPluginClient{applyResponse: &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{ {EventId: testWatchHistoryID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED}, { EventId: testSecondHistoryID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_RETRY, Fault: &pluginv1.WatchSyncFault{ Code: pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_RATE_LIMITED, SafeMessage: "slow down", RetryAfter: durationpb.New(45 * time.Second), }, }, }}} provider := testPluginProvider(t, client) result, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{ {HistoryID: testWatchHistoryID, Kind: historyimport.KindMovie}, {HistoryID: testSecondHistoryID, Kind: historyimport.KindMovie}, {HistoryID: "history-3", Kind: historyimport.KindMovie}, }) limited, ok := AsRateLimited(err) if !ok || limited.RetryAfter != 45*time.Second { t.Fatalf("error = %#v", err) } if len(result.Sent) != 1 || result.Sent[0] != testWatchHistoryID || len(result.Failed) != 0 { t.Fatalf("result = %#v", result) } } func TestPluginProviderTransportFailureIsRetryableAndSanitized(t *testing.T) { client := &fakeWatchSyncPluginClient{applyErr: errors.New("rpc failed with access_token=secret")} provider := testPluginProvider(t, client) _, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{{HistoryID: testWatchHistoryID, Kind: historyimport.KindMovie}}) if !isRetryableProviderError(err) || strings.Contains(err.Error(), testSecretValue) { t.Fatalf("error = %#v", err) } } func TestPluginProviderSanitizesFaultMessage(t *testing.T) { message := safeApplyMessage(&pluginv1.WatchSyncApplyResult{Fault: &pluginv1.WatchSyncFault{ SafeMessage: " failed\n\taccess-token " + strings.Repeat("x", 300), }}, "access-token") if strings.ContainsAny(message, "\n\t") || strings.Contains(message, "access-token") || len([]rune(message)) > 257 { t.Fatalf("message was not sanitized: %q", message) } } func TestPluginProviderNormalizesSecretsBeforeRedaction(t *testing.T) { message := safeApplyMessage(&pluginv1.WatchSyncApplyResult{Fault: &pluginv1.WatchSyncFault{ SafeMessage: "credential line one line two was rejected", }}, "line one\nline two") if strings.Contains(message, "line one line two") || !strings.Contains(message, "[REDACTED]") { t.Fatalf("message was not redacted: %q", message) } } func TestPluginProviderMapsTemporaryRetryToFailed(t *testing.T) { client := &fakeWatchSyncPluginClient{applyResponse: &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: testWatchHistoryID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_RETRY, Fault: &pluginv1.WatchSyncFault{ Code: pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_TEMPORARY, SafeMessage: "temporary upstream failure", }, }}}} provider := testPluginProvider(t, client) result, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{{HistoryID: testWatchHistoryID, Kind: historyimport.KindMovie}}) if err != nil { t.Fatal(err) } if result.Failed[testWatchHistoryID] != "temporary upstream failure" { t.Fatalf("result = %#v", result) } } func TestPluginProviderMapsRateLimitFault(t *testing.T) { client := &fakeWatchSyncPluginClient{applyResponse: &pluginv1.WatchSyncApplyEventsResponse{Fault: &pluginv1.WatchSyncFault{ Code: pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_RATE_LIMITED, RetryAfter: durationpb.New(30 * time.Second), }}} provider := testPluginProvider(t, client) _, err := provider.ExportHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{{HistoryID: "h", ProviderItemKey: "p", Kind: historyimport.KindMovie}}) limited, ok := AsRateLimited(err) if !ok || limited.RetryAfter != 30*time.Second { t.Fatalf("error = %#v", err) } } func TestPluginProviderRejectsUnsupportedScrobbleMedia(t *testing.T) { client := &fakeWatchSyncPluginClient{} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ExportWatched: true, MaxBatchSize: 25, SupportedMediaTypes: []pluginv1.WatchSyncMediaType{pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE}, }) err := provider.Stop(context.Background(), ServerConfig{}, Connection{}, ScrobbleEvent{ Completed: true, HistoryID: testWatchHistoryID, PlaybackSessionID: testPlaybackSessionID, Kind: historyimport.KindEpisode, OccurredAt: time.Now().UTC(), }) if err == nil || err.Error() != watchSyncUnsupportedEpisodeMediaMessage { t.Fatalf("error = %#v", err) } if client.applyRequest != nil { t.Fatalf("unexpected apply request = %#v", client.applyRequest) } } func TestPluginProviderPreservesScrobbleRetryClassification(t *testing.T) { client := &fakeWatchSyncPluginClient{} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ScrobblePlayback: true, MaxBatchSize: 25, }) event := ScrobbleEvent{ PlaybackSessionID: testPlaybackSessionID, MediaItemID: testMovieMediaID, Kind: historyimport.KindMovie, OccurredAt: time.Now().UTC(), } eventID := "scrobble:" + pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_START.String() + ":" + testPlaybackSessionID client.applyResponse = &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: eventID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_RETRY, Fault: &pluginv1.WatchSyncFault{ Code: pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_TEMPORARY, SafeMessage: "retry later", }, }}} if err := provider.Start(context.Background(), ServerConfig{}, Connection{}, event); !isRetryableProviderError(err) { t.Fatalf("retry error = %#v, want retryableProviderError", err) } client.applyResponse = &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: eventID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_REJECTED, }}} if err := provider.Start(context.Background(), ServerConfig{}, Connection{}, event); err == nil || isRetryableProviderError(err) { t.Fatalf("rejected error = %#v, want terminal error", err) } } func TestPluginProviderAuthenticatedContextUsesCapabilityAndCredentials(t *testing.T) { client := &fakeWatchSyncPluginClient{} provider := testPluginProvider(t, client) if _, err := provider.LookupAccount(context.Background(), ServerConfig{}, Connection{AccessToken: "token"}); err != nil { t.Fatal(err) } if client.accountRequest.GetContext().GetCapabilityId() != testPluginCapabilityID || client.accountRequest.GetContext().GetCredentials().GetAccessToken() != "token" { t.Fatalf("account request = %#v", client.accountRequest) } _ = testPlaybackSessionID } func TestPluginProviderSupportsDeviceAuthorizationAndFullCredentials(t *testing.T) { expiresAt := time.Now().UTC().Add(10 * time.Minute) client := &fakeWatchSyncPluginClient{ deviceStartResponse: &pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse{ UserCode: "ABCD", VerificationUrl: "https://provider.example/activate", VerificationUrlComplete: "https://provider.example/activate?code=ABCD", ProviderState: []byte("opaque-device-state"), PollingInterval: durationpb.New(7 * time.Second), ExpiresAt: timestamppb.New(expiresAt), }, devicePollResponse: &pluginv1.WatchSyncDeviceAuthorizationServicePollResponse{ Status: pluginv1.WatchSyncDeviceAuthorizationStatus_WATCH_SYNC_DEVICE_AUTHORIZATION_STATUS_AUTHORIZED, Credentials: &pluginv1.WatchSyncCredentials{ AccessToken: testAccessToken, RefreshToken: testRefreshToken, TokenType: testDPoPTokenType, Scopes: []string{testHistoryScope, "watchlist"}, SecretAttributes: map[string]string{"instance": testOneValue}, ExpiresAt: timestamppb.New(expiresAt), }, }, } provider, err := NewPluginProvider(PluginProviderOptions{ InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID, Descriptor: &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_DEVICE_CODE}, ImportWatched: true, MaxBatchSize: 25, }, ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) { return client, nil }, ResolveConfig: func(context.Context, int) (*pluginv1.WatchSyncProviderConfig, error) { return &pluginv1.WatchSyncProviderConfig{SecretValues: map[string]string{"provider.client_secret": testSecretValue}}, nil }, }) if err != nil { t.Fatal(err) } session, err := provider.StartDeviceAuth(context.Background(), ServerConfig{}) if err != nil { t.Fatal(err) } if session.UserCode != "ABCD" || session.IntervalSeconds != 7 || session.VerificationURL != "https://provider.example/activate?code=ABCD" || client.deviceStartRequest.GetProviderConfig().GetSecretValues()["provider.client_secret"] != testSecretValue { t.Fatalf("session=%#v request=%#v", session, client.deviceStartRequest) } tokens, err := provider.PollDeviceAuth(context.Background(), ServerConfig{}, session) if err != nil { t.Fatal(err) } if string(client.devicePollRequest.GetProviderState()) != "opaque-device-state" || tokens.TokenType != testDPoPTokenType || len(tokens.Scopes) != 2 || tokens.SecretAttributes["instance"] != testOneValue { t.Fatalf("tokens=%#v request=%#v", tokens, client.devicePollRequest) } } func TestPluginProviderRejectsInvalidDeviceAuthorizationMetadata(t *testing.T) { valid := func() *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse { return &pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse{ UserCode: "ABCD", VerificationUrl: "https://provider.example/activate", ProviderState: []byte("opaque"), PollingInterval: durationpb.New(5 * time.Second), ExpiresAt: timestamppb.New(time.Now().UTC().Add(10 * time.Minute)), } } tests := map[string]func(*pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse){ "relative URL": func(response *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse) { response.VerificationUrl = "/activate" }, "URL userinfo": func(response *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse) { response.VerificationUrl = "https://user:pass@provider.example/activate" }, "unsafe complete URL": func(response *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse) { response.VerificationUrlComplete = "javascript:alert(1)" }, "invalid timestamp": func(response *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse) { response.ExpiresAt = ×tamppb.Timestamp{Seconds: 253402300800} }, "expired timestamp": func(response *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse) { response.ExpiresAt = timestamppb.New(time.Now().UTC().Add(-time.Minute)) }, "invalid interval": func(response *pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse) { response.PollingInterval = durationpb.New(-time.Second) }, } for name, mutate := range tests { t.Run(name, func(t *testing.T) { response := valid() mutate(response) client := &fakeWatchSyncPluginClient{deviceStartResponse: response} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_DEVICE_CODE}, }) if _, err := provider.StartDeviceAuth(context.Background(), ServerConfig{}); err == nil { t.Fatal("StartDeviceAuth error = nil") } }) } } func TestPluginProviderPendingDeviceAuthorizationCarriesRotatedState(t *testing.T) { expiresAt := time.Now().UTC().Add(20 * time.Minute) client := &fakeWatchSyncPluginClient{devicePollResponse: &pluginv1.WatchSyncDeviceAuthorizationServicePollResponse{ Status: pluginv1.WatchSyncDeviceAuthorizationStatus_WATCH_SYNC_DEVICE_AUTHORIZATION_STATUS_PENDING, ProviderState: []byte("rotated-state"), PollingInterval: durationpb.New(11 * time.Second), ExpiresAt: timestamppb.New(expiresAt), }} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_DEVICE_CODE}, }) original := DeviceAuthSession{ ID: "auth-1", Provider: testPluginProviderKey, UserID: 7, ProfileID: "profile-1", DeviceCode: base64.RawURLEncoding.EncodeToString([]byte("original-state")), UserCode: "ABCD", VerificationURL: "https://provider.example/activate", IntervalSeconds: 5, ExpiresAt: time.Now().UTC().Add(10 * time.Minute), } _, err := provider.PollDeviceAuth(context.Background(), ServerConfig{}, original) var pending deviceAuthorizationPendingError if !errors.As(err, &pending) { t.Fatalf("error = %#v, want deviceAuthorizationPendingError", err) } if pending.session.DeviceCode != base64.RawURLEncoding.EncodeToString([]byte("rotated-state")) || pending.session.IntervalSeconds != 11 || !pending.session.ExpiresAt.Equal(expiresAt) { t.Fatalf("pending session = %#v", pending.session) } } func TestPluginProviderPendingDeviceAuthorizationPreservesStatePresence(t *testing.T) { originalState := base64.RawURLEncoding.EncodeToString([]byte("original-state")) tests := map[string]struct { providerState []byte wantState string }{ "omitted retains state": {providerState: nil, wantState: originalState}, "explicit empty clears state": { providerState: []byte{}, wantState: "", }, } for name, test := range tests { t.Run(name, func(t *testing.T) { client := &fakeWatchSyncPluginClient{devicePollResponse: &pluginv1.WatchSyncDeviceAuthorizationServicePollResponse{ Status: pluginv1.WatchSyncDeviceAuthorizationStatus_WATCH_SYNC_DEVICE_AUTHORIZATION_STATUS_PENDING, ProviderState: test.providerState, }} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{ pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_DEVICE_CODE, }, }) original := DeviceAuthSession{ ID: "auth-1", Provider: testPluginProviderKey, UserID: 7, ProfileID: "profile-1", DeviceCode: originalState, UserCode: "ABCD", VerificationURL: "https://provider.example/activate", IntervalSeconds: 5, ExpiresAt: time.Now().UTC().Add(10 * time.Minute), } _, err := provider.PollDeviceAuth(context.Background(), ServerConfig{}, original) var pending deviceAuthorizationPendingError if !errors.As(err, &pending) { t.Fatalf("error = %#v, want deviceAuthorizationPendingError", err) } if pending.session.DeviceCode != test.wantState { t.Fatalf("device state = %q, want %q", pending.session.DeviceCode, test.wantState) } }) } } func TestPluginProviderPaginatesRemoteStateAndPersistsRotatedCredentials(t *testing.T) { now := time.Now().UTC() remote := func(key, imdb string) *pluginv1.WatchSyncRemoteState { return &pluginv1.WatchSyncRemoteState{ ProviderItemKey: key, Media: &pluginv1.WatchSyncMedia{ MediaType: pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE, Title: "Movie", ExternalIds: map[string]string{"imdb": imdb}, }, Watched: &pluginv1.WatchSyncRemoteWatchedState{PlayCount: 1, LastWatchedAt: timestamppb.New(now)}, } } client := &fakeWatchSyncPluginClient{listResponses: []*pluginv1.WatchSyncListRemoteStateResponse{ { Items: []*pluginv1.WatchSyncRemoteState{remote(testOneValue, "tt1")}, NextPageToken: "page-2", CompleteSnapshot: true, UpdatedCredentials: &pluginv1.WatchSyncCredentials{ AccessToken: "rotated", TokenType: testBearerTokenType, Scopes: []string{testHistoryScope}, }, }, {Items: []*pluginv1.WatchSyncRemoteState{remote("two", "tt2")}, NextCursor: "cursor-2", CompleteSnapshot: true}, }} repository := &fakePluginCredentialRepository{} provider, err := NewPluginProvider(PluginProviderOptions{ InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID, Descriptor: &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ImportWatched: true, MaxBatchSize: 25, }, ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) { return client, nil }, Repository: repository, }) if err != nil { t.Fatal(err) } batch, err := provider.FetchWatchedBatch(context.Background(), ServerConfig{}, Connection{ ID: "connection", Provider: testPluginProviderKey, UserID: 1, ProfileID: "profile", AccessToken: "old", SyncCursors: map[string]string{pluginWatchedCursorKey: testCursorOne}, }) if err != nil { t.Fatal(err) } if len(batch.Rows) != 2 || batch.UpdatedCursors[pluginWatchedCursorKey] != "cursor-2" || len(client.listRequests) != 2 || client.listRequests[0].GetCursor() != testCursorOne || client.listRequests[1].GetCursor() != testCursorOne || client.listRequests[1].GetPageToken() != "page-2" { t.Fatalf("batch=%#v requests=%#v", batch, client.listRequests) } if repository.saved.AccessToken != "rotated" || repository.saved.Scopes[0] != testHistoryScope { t.Fatalf("persisted credentials = %#v", repository.saved) } } func TestPluginProviderBoundsRemoteStateTraversalByItemCount(t *testing.T) { client := &fakeWatchSyncPluginClient{listResponses: []*pluginv1.WatchSyncListRemoteStateResponse{ {Items: make([]*pluginv1.WatchSyncRemoteState, maxRemoteStateItems), NextPageToken: "page-2"}, {Items: []*pluginv1.WatchSyncRemoteState{{}}, NextCursor: "must-not-commit"}, }} provider := testPluginProvider(t, client) batch, err := provider.FetchWatchedBatch(context.Background(), ServerConfig{}, Connection{ SyncCursors: map[string]string{pluginWatchedCursorKey: testCursorOne}, }) if err == nil || !strings.Contains(err.Error(), "item limit") { t.Fatalf("error = %v, want item limit", err) } if len(batch.UpdatedCursors) != 0 || len(client.listRequests) != 2 || client.listRequests[1].GetCursor() != testCursorOne { t.Fatalf("batch=%#v requests=%#v", batch, client.listRequests) } } func TestPluginProviderRejectsIncrementalOrderedWatchlist(t *testing.T) { client := &fakeWatchSyncPluginClient{listResponse: &pluginv1.WatchSyncListRemoteStateResponse{ NextCursor: "must-not-commit", CompleteSnapshot: false, }} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ImportWatchlist: true, ProvidesWatchlistOrder: true, MaxBatchSize: 25, }) batch, err := provider.FetchWatchlistBatch(context.Background(), ServerConfig{}, Connection{ SyncCursors: map[string]string{pluginWatchlistCursorKey: testCursorOne}, }) if err == nil || !strings.Contains(err.Error(), "incremental traversal for an ordered watchlist") { t.Fatalf("error = %v, want ordered watchlist snapshot requirement", err) } if len(batch.UpdatedCursors) != 0 || len(client.listRequests) != 1 { t.Fatalf("batch=%#v requests=%#v", batch, client.listRequests) } } func TestPluginProviderDecodesKeyOnlyListTombstone(t *testing.T) { row, err := remoteFavoriteFromProto(testPluginProviderKey, &pluginv1.WatchSyncRemoteState{ ProviderItemKey: "remote-1", }, &pluginv1.WatchSyncRemoteListState{Removed: true}) if err != nil { t.Fatal(err) } if !row.Removed || row.ProviderItemKey != "remote-1" || row.Kind != "" { t.Fatalf("row = %#v", row) } } func TestPluginProviderMapsAllCapabilitiesAndListOperations(t *testing.T) { descriptor := &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ImportWatched: true, ImportProgress: true, ExportWatched: true, ExportUnwatched: true, ImportFavorites: true, ExportFavorites: true, RemoveFavorites: true, ImportWatchlist: true, ExportWatchlist: true, RemoveWatchlist: true, ProvidesWatchlistOrder: true, ScrobblePlayback: true, MaxBatchSize: 25, } client := &fakeWatchSyncPluginClient{} provider := testPluginProviderWithDescriptor(t, client, descriptor) if provider.Capabilities() != (Capabilities{ ImportWatched: true, ImportProgress: true, ExportWatched: true, ExportUnwatched: true, ImportFavorites: true, ExportFavorites: true, RemoveFavorites: true, ImportWatchlist: true, ExportWatchlist: true, RemoveWatchlist: true, ProvidesWatchlistOrder: true, ScrobblePlayback: true, }) { t.Fatalf("capabilities = %#v", provider.Capabilities()) } item := LocalFavorite{MediaItemID: testMovieMediaID, ProviderItemKey: "remote-1", Kind: historyimport.KindMovie, IMDbID: "tt1"} eventID := pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_ADD_TO_WATCHLIST.String() + ":movie-1" client.applyResponse = &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: eventID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED, }}} result, err := provider.ExportWatchlist(context.Background(), ServerConfig{}, Connection{}, []LocalFavorite{item}) if err != nil { t.Fatal(err) } event := client.applyRequest.GetEvents()[0] if len(result.Sent) != 1 || event.GetOperation() != pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_ADD_TO_WATCHLIST || event.GetProviderItemKey() != "remote-1" { t.Fatalf("result=%#v event=%#v", result, event) } } func TestPluginProviderListEventsKeepFailuresAndPresenceAwareOrder(t *testing.T) { client := &fakeWatchSyncPluginClient{} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ExportWatchlist: true, RemoveWatchlist: true, SupportedMediaTypes: []pluginv1.WatchSyncMediaType{pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE}, MaxBatchSize: 25, }) items := []LocalFavorite{ {MediaItemID: testEpisodeMediaID, Kind: historyimport.KindEpisode}, {MediaItemID: testMovieMediaID, Kind: historyimport.KindMovie}, } addEventID := pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_ADD_TO_WATCHLIST.String() + ":" + testMovieMediaID client.applyResponse = &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: addEventID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED, }}} result, err := provider.ExportWatchlist(context.Background(), ServerConfig{}, Connection{}, items) if err != nil { t.Fatal(err) } event := client.applyRequest.GetEvents()[0] if result.Failed[testEpisodeMediaID] != watchSyncUnsupportedEpisodeMediaMessage || event.ListPosition == nil || event.GetListPosition() != 0 { t.Fatalf("result=%#v event=%#v", result, event) } removeEventID := pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_REMOVE_FROM_WATCHLIST.String() + ":" + testMovieMediaID client.applyResponse = &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: removeEventID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED, }}} result, err = provider.RemoveWatchlist(context.Background(), ServerConfig{}, Connection{}, items) if err != nil { t.Fatal(err) } event = client.applyRequest.GetEvents()[0] if result.Failed[testEpisodeMediaID] != watchSyncUnsupportedEpisodeMediaMessage || event.ListPosition != nil { t.Fatalf("result=%#v event=%#v", result, event) } } func TestPluginProviderRemoveHistoryRecordsUnsupportedMedia(t *testing.T) { client := &fakeWatchSyncPluginClient{} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ExportUnwatched: true, SupportedMediaTypes: []pluginv1.WatchSyncMediaType{pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE}, }) result, err := provider.RemoveHistory(context.Background(), ServerConfig{}, Connection{}, []LocalPlay{{ HistoryID: testWatchHistoryID, Kind: historyimport.KindEpisode, }}) if err != nil { t.Fatal(err) } if result.Failed[testWatchHistoryID] != watchSyncUnsupportedEpisodeMediaMessage || client.applyRequest != nil { t.Fatalf("result=%#v request=%#v", result, client.applyRequest) } } func TestPluginProviderForwardsLiveScrobbleLifecycle(t *testing.T) { client := &fakeWatchSyncPluginClient{} provider := testPluginProviderWithDescriptor(t, client, &pluginv1.WatchSyncProviderDescriptor{ AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY}, ScrobblePlayback: true, MaxBatchSize: 25, }) event := ScrobbleEvent{PlaybackSessionID: testPlaybackSessionID, MediaItemID: testMovieMediaID, Kind: historyimport.KindMovie, PositionSeconds: 12.5} eventID := "scrobble:" + pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_START.String() + ":" + testPlaybackSessionID client.applyResponse = &pluginv1.WatchSyncApplyEventsResponse{Results: []*pluginv1.WatchSyncApplyResult{{ EventId: eventID, Status: pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED, }}} if err := provider.Start(context.Background(), ServerConfig{}, Connection{}, event); err != nil { t.Fatal(err) } if got := client.applyRequest.GetEvents()[0]; got.GetOperation() != pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_START || got.GetPositionSeconds() != 12.5 { t.Fatalf("scrobble event = %#v", got) } }