Merge pull request #117 from NicolasConstant/topic_remove-deleted-twitter-users

Topic remove deleted twitter users
This commit is contained in:
Nicolas Constant 2021-09-10 01:15:54 -04:00 committed by GitHub
commit 363481a997
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
25 changed files with 532 additions and 74 deletions

View file

@ -13,6 +13,7 @@
<ItemGroup>
<ProjectReference Include="..\BirdsiteLive.Domain\BirdsiteLive.Domain.csproj" />
<ProjectReference Include="..\BirdsiteLive.Moderation\BirdsiteLive.Moderation.csproj" />
<ProjectReference Include="..\BirdsiteLive.Twitter\BirdsiteLive.Twitter.csproj" />
<ProjectReference Include="..\DataAccessLayers\BirdsiteLive.DAL\BirdsiteLive.DAL.csproj" />
</ItemGroup>

View file

@ -0,0 +1,12 @@
using System.Threading;
using System.Threading.Tasks;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Pipeline.Models;
namespace BirdsiteLive.Pipeline.Contracts
{
public interface IRefreshTwitterUserStatusProcessor
{
Task<UserWithDataToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct);
}
}

View file

@ -7,7 +7,7 @@ namespace BirdsiteLive.Pipeline.Contracts
{
public interface IRetrieveFollowersProcessor
{
Task<IEnumerable<UserWithTweetsToSync>> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct);
Task<IEnumerable<UserWithDataToSync>> ProcessAsync(UserWithDataToSync[] userWithTweetsToSyncs, CancellationToken ct);
//IAsyncEnumerable<UserWithTweetsToSync> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct);
}
}

View file

@ -7,6 +7,6 @@ namespace BirdsiteLive.Pipeline.Contracts
{
public interface IRetrieveTweetsProcessor
{
Task<UserWithTweetsToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct);
Task<UserWithDataToSync[]> ProcessAsync(UserWithDataToSync[] syncTwitterUsers, CancellationToken ct);
}
}

View file

@ -6,6 +6,6 @@ namespace BirdsiteLive.Pipeline.Contracts
{
public interface ISaveProgressionProcessor
{
Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct);
Task ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct);
}
}

View file

@ -6,6 +6,6 @@ namespace BirdsiteLive.Pipeline.Contracts
{
public interface ISendTweetsToFollowersProcessor
{
Task<UserWithTweetsToSync> ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct);
Task<UserWithDataToSync> ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct);
}
}

View file

@ -4,7 +4,7 @@ using Tweetinvi.Models;
namespace BirdsiteLive.Pipeline.Models
{
public class UserWithTweetsToSync
public class UserWithDataToSync
{
public SyncTwitterUser User { get; set; }
public ExtractedTweet[] Tweets { get; set; }

View file

@ -0,0 +1,72 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Moderation.Actions;
using BirdsiteLive.Pipeline.Contracts;
using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Twitter;
namespace BirdsiteLive.Pipeline.Processors
{
public class RefreshTwitterUserStatusProcessor : IRefreshTwitterUserStatusProcessor
{
private const int FetchingErrorCountThreshold = 300;
private readonly ICachedTwitterUserService _twitterUserService;
private readonly ITwitterUserDal _twitterUserDal;
private readonly IRemoveTwitterAccountAction _removeTwitterAccountAction;
#region Ctor
public RefreshTwitterUserStatusProcessor(ICachedTwitterUserService twitterUserService, ITwitterUserDal twitterUserDal, IRemoveTwitterAccountAction removeTwitterAccountAction)
{
_twitterUserService = twitterUserService;
_twitterUserDal = twitterUserDal;
_removeTwitterAccountAction = removeTwitterAccountAction;
}
#endregion
public async Task<UserWithDataToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct)
{
var usersWtData = new List<UserWithDataToSync>();
foreach (var user in syncTwitterUsers)
{
var userView = _twitterUserService.GetUser(user.Acct);
if (userView == null)
{
await AnalyseFailingUserAsync(user);
}
else if (!userView.Protected)
{
user.FetchingErrorCount = 0;
var userWtData = new UserWithDataToSync
{
User = user
};
usersWtData.Add(userWtData);
}
}
return usersWtData.ToArray();
}
private async Task AnalyseFailingUserAsync(SyncTwitterUser user)
{
var dbUser = await _twitterUserDal.GetTwitterUserAsync(user.Acct);
dbUser.FetchingErrorCount++;
if (dbUser.FetchingErrorCount > FetchingErrorCountThreshold)
{
await _removeTwitterAccountAction.ProcessAsync(user);
}
else
{
await _twitterUserDal.UpdateTwitterUserAsync(dbUser);
}
// Purge
_twitterUserService.PurgeUser(user.Acct);
}
}
}

View file

@ -18,7 +18,7 @@ namespace BirdsiteLive.Pipeline.Processors
}
#endregion
public async Task<IEnumerable<UserWithTweetsToSync>> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct)
public async Task<IEnumerable<UserWithDataToSync>> ProcessAsync(UserWithDataToSync[] userWithTweetsToSyncs, CancellationToken ct)
{
//TODO multithread this
foreach (var user in userWithTweetsToSyncs)

View file

@ -31,33 +31,30 @@ namespace BirdsiteLive.Pipeline.Processors
}
#endregion
public async Task<UserWithTweetsToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct)
public async Task<UserWithDataToSync[]> ProcessAsync(UserWithDataToSync[] syncTwitterUsers, CancellationToken ct)
{
var usersWtTweets = new List<UserWithTweetsToSync>();
var usersWtTweets = new List<UserWithDataToSync>();
//TODO multithread this
foreach (var user in syncTwitterUsers)
foreach (var userWtData in syncTwitterUsers)
{
var user = userWtData.User;
var tweets = RetrieveNewTweets(user);
if (tweets.Length > 0 && user.LastTweetPostedId != -1)
{
var userWtTweets = new UserWithTweetsToSync
{
User = user,
Tweets = tweets
};
usersWtTweets.Add(userWtTweets);
userWtData.Tweets = tweets;
usersWtTweets.Add(userWtData);
}
else if (tweets.Length > 0 && user.LastTweetPostedId == -1)
{
var tweetId = tweets.Last().Id;
var now = DateTime.UtcNow;
await _twitterUserDal.UpdateTwitterUserAsync(user.Id, tweetId, tweetId, now);
await _twitterUserDal.UpdateTwitterUserAsync(user.Id, tweetId, tweetId, user.FetchingErrorCount, now);
}
else
{
var now = DateTime.UtcNow;
await _twitterUserDal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, now);
await _twitterUserDal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, user.FetchingErrorCount, now);
}
}
@ -67,11 +64,7 @@ namespace BirdsiteLive.Pipeline.Processors
private ExtractedTweet[] RetrieveNewTweets(SyncTwitterUser user)
{
var tweets = new ExtractedTweet[0];
// Don't retrieve TL if protected
var userView = _twitterUserService.GetUser(user.Acct);
if (userView == null || userView.Protected) return tweets;
try
{
if (user.LastTweetPostedId == -1)

View file

@ -22,7 +22,7 @@ namespace BirdsiteLive.Pipeline.Processors
}
#endregion
public async Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct)
public async Task ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct)
{
try
{
@ -49,7 +49,7 @@ namespace BirdsiteLive.Pipeline.Processors
var lastPostedTweet = userWithTweetsToSync.Tweets.Select(x => x.Id).Max();
var minimumSync = followingSyncStatuses.Min();
var now = DateTime.UtcNow;
await _twitterUserDal.UpdateTwitterUserAsync(userId, lastPostedTweet, minimumSync, now);
await _twitterUserDal.UpdateTwitterUserAsync(userId, lastPostedTweet, minimumSync, userWithTweetsToSync.User.FetchingErrorCount, now);
}
catch (Exception e)
{

View file

@ -33,7 +33,7 @@ namespace BirdsiteLive.Pipeline.Processors
}
#endregion
public async Task<UserWithTweetsToSync> ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct)
public async Task<UserWithDataToSync> ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct)
{
var user = userWithTweetsToSync.User;

View file

@ -1,4 +1,5 @@
using System;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
@ -17,6 +18,7 @@ namespace BirdsiteLive.Pipeline
public class StatusPublicationPipeline : IStatusPublicationPipeline
{
private readonly IRetrieveTwitterUsersProcessor _retrieveTwitterAccountsProcessor;
private readonly IRefreshTwitterUserStatusProcessor _refreshTwitterUserStatusProcessor;
private readonly IRetrieveTweetsProcessor _retrieveTweetsProcessor;
private readonly IRetrieveFollowersProcessor _retrieveFollowersProcessor;
private readonly ISendTweetsToFollowersProcessor _sendTweetsToFollowersProcessor;
@ -24,13 +26,14 @@ namespace BirdsiteLive.Pipeline
private readonly ILogger<StatusPublicationPipeline> _logger;
#region Ctor
public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, ILogger<StatusPublicationPipeline> logger)
public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, IRefreshTwitterUserStatusProcessor refreshTwitterUserStatusProcessor, ILogger<StatusPublicationPipeline> logger)
{
_retrieveTweetsProcessor = retrieveTweetsProcessor;
_retrieveTwitterAccountsProcessor = retrieveTwitterAccountsProcessor;
_retrieveFollowersProcessor = retrieveFollowersProcessor;
_sendTweetsToFollowersProcessor = sendTweetsToFollowersProcessor;
_saveProgressionProcessor = saveProgressionProcessor;
_refreshTwitterUserStatusProcessor = refreshTwitterUserStatusProcessor;
_logger = logger;
}
@ -39,16 +42,21 @@ namespace BirdsiteLive.Pipeline
public async Task ExecuteAsync(CancellationToken ct)
{
// Create blocks
var twitterUsersBufferBlock = new BufferBlock<SyncTwitterUser[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
var retrieveTweetsBlock = new TransformBlock<SyncTwitterUser[], UserWithTweetsToSync[]>(async x => await _retrieveTweetsProcessor.ProcessAsync(x, ct));
var retrieveTweetsBufferBlock = new BufferBlock<UserWithTweetsToSync[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
var retrieveFollowersBlock = new TransformManyBlock<UserWithTweetsToSync[], UserWithTweetsToSync>(async x => await _retrieveFollowersProcessor.ProcessAsync(x, ct));
var retrieveFollowersBufferBlock = new BufferBlock<UserWithTweetsToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
var sendTweetsToFollowersBlock = new TransformBlock<UserWithTweetsToSync, UserWithTweetsToSync>(async x => await _sendTweetsToFollowersProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
var sendTweetsToFollowersBufferBlock = new BufferBlock<UserWithTweetsToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
var saveProgressionBlock = new ActionBlock<UserWithTweetsToSync>(async x => await _saveProgressionProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
var twitterUserToRefreshBufferBlock = new BufferBlock<SyncTwitterUser[]>(new DataflowBlockOptions
{ BoundedCapacity = 1, CancellationToken = ct });
var twitterUserToRefreshBlock = new TransformBlock<SyncTwitterUser[], UserWithDataToSync[]>(async x => await _refreshTwitterUserStatusProcessor.ProcessAsync(x, ct));
var twitterUsersBufferBlock = new BufferBlock<UserWithDataToSync[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
var retrieveTweetsBlock = new TransformBlock<UserWithDataToSync[], UserWithDataToSync[]>(async x => await _retrieveTweetsProcessor.ProcessAsync(x, ct));
var retrieveTweetsBufferBlock = new BufferBlock<UserWithDataToSync[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
var retrieveFollowersBlock = new TransformManyBlock<UserWithDataToSync[], UserWithDataToSync>(async x => await _retrieveFollowersProcessor.ProcessAsync(x, ct));
var retrieveFollowersBufferBlock = new BufferBlock<UserWithDataToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
var sendTweetsToFollowersBlock = new TransformBlock<UserWithDataToSync, UserWithDataToSync>(async x => await _sendTweetsToFollowersProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
var sendTweetsToFollowersBufferBlock = new BufferBlock<UserWithDataToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
var saveProgressionBlock = new ActionBlock<UserWithDataToSync>(async x => await _saveProgressionProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
// Link pipeline
twitterUserToRefreshBufferBlock.LinkTo(twitterUserToRefreshBlock, new DataflowLinkOptions { PropagateCompletion = true });
twitterUserToRefreshBlock.LinkTo(twitterUsersBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
twitterUsersBufferBlock.LinkTo(retrieveTweetsBlock, new DataflowLinkOptions { PropagateCompletion = true });
retrieveTweetsBlock.LinkTo(retrieveTweetsBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
retrieveTweetsBufferBlock.LinkTo(retrieveFollowersBlock, new DataflowLinkOptions { PropagateCompletion = true });
@ -58,7 +66,7 @@ namespace BirdsiteLive.Pipeline
sendTweetsToFollowersBufferBlock.LinkTo(saveProgressionBlock, new DataflowLinkOptions { PropagateCompletion = true });
// Launch twitter user retriever
var retrieveTwitterAccountsTask = _retrieveTwitterAccountsProcessor.GetTwitterUsersAsync(twitterUsersBufferBlock, ct);
var retrieveTwitterAccountsTask = _retrieveTwitterAccountsProcessor.GetTwitterUsersAsync(twitterUserToRefreshBufferBlock, ct);
// Wait
await Task.WhenAny(new[] { retrieveTwitterAccountsTask, saveProgressionBlock.Completion });

View file

@ -49,6 +49,9 @@ namespace BirdsiteLive.Twitter
catch (Exception e)
{
_logger.LogError(e, "Error retrieving user {Username}", username);
// TODO keep track of error, see where to remove user if too much errors
return null;
}

View file

@ -23,7 +23,7 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
public class DbInitializerPostgresDal : PostgresBase, IDbInitializerDal
{
private readonly PostgresTools _tools;
private readonly Version _currentVersion = new Version(2, 1);
private readonly Version _currentVersion = new Version(2, 2);
private const string DbVersionType = "db-version";
#region Ctor
@ -132,7 +132,8 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
return new[]
{
new Tuple<Version, Version>(new Version(1,0), new Version(2,0)),
new Tuple<Version, Version>(new Version(2,0), new Version(2,1))
new Tuple<Version, Version>(new Version(2,0), new Version(2,1)),
new Tuple<Version, Version>(new Version(2,1), new Version(2,2))
};
}
@ -151,6 +152,11 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
var addActorId = $@"ALTER TABLE {_settings.FollowersTableName} ADD actorId VARCHAR(2048)";
await _tools.ExecuteRequestAsync(addActorId);
}
else if (from == new Version(2, 1) && to == new Version(2, 2))
{
var addLastSync = $@"ALTER TABLE {_settings.TwitterUserTableName} ADD fetchingErrorCount SMALLINT";
await _tools.ExecuteRequestAsync(addLastSync);
}
else
{
throw new NotImplementedException();

View file

@ -99,23 +99,28 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
}
}
public async Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, DateTime lastSync)
public async Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, int fetchingErrorCount, DateTime lastSync)
{
if(id == default) throw new ArgumentException("id");
if(lastTweetPostedId == default) throw new ArgumentException("lastTweetPostedId");
if(lastTweetSynchronizedForAllFollowersId == default) throw new ArgumentException("lastTweetSynchronizedForAllFollowersId");
if(lastSync == default) throw new ArgumentException("lastSync");
var query = $"UPDATE {_settings.TwitterUserTableName} SET lastTweetPostedId = @lastTweetPostedId, lastTweetSynchronizedForAllFollowersId = @lastTweetSynchronizedForAllFollowersId, lastSync = @lastSync WHERE id = @id";
var query = $"UPDATE {_settings.TwitterUserTableName} SET lastTweetPostedId = @lastTweetPostedId, lastTweetSynchronizedForAllFollowersId = @lastTweetSynchronizedForAllFollowersId, fetchingErrorCount = @fetchingErrorCount, lastSync = @lastSync WHERE id = @id";
using (var dbConnection = Connection)
{
dbConnection.Open();
await dbConnection.QueryAsync(query, new { id, lastTweetPostedId, lastTweetSynchronizedForAllFollowersId, lastSync = lastSync.ToUniversalTime() });
await dbConnection.QueryAsync(query, new { id, lastTweetPostedId, lastTweetSynchronizedForAllFollowersId, fetchingErrorCount, lastSync = lastSync.ToUniversalTime() });
}
}
public async Task UpdateTwitterUserAsync(SyncTwitterUser user)
{
await UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, user.FetchingErrorCount, user.LastSync);
}
public async Task DeleteTwitterUserAsync(string acct)
{
if (string.IsNullOrWhiteSpace(acct)) throw new ArgumentException("acct");

View file

@ -11,7 +11,8 @@ namespace BirdsiteLive.DAL.Contracts
Task<SyncTwitterUser> GetTwitterUserAsync(int id);
Task<SyncTwitterUser[]> GetAllTwitterUsersAsync(int maxNumber);
Task<SyncTwitterUser[]> GetAllTwitterUsersAsync();
Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, DateTime lastSync);
Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, int fetchingErrorCount, DateTime lastSync);
Task UpdateTwitterUserAsync(SyncTwitterUser user);
Task DeleteTwitterUserAsync(string acct);
Task DeleteTwitterUserAsync(int id);
Task<int> GetTwitterUsersCountAsync();

View file

@ -11,5 +11,7 @@ namespace BirdsiteLive.DAL.Models
public long LastTweetSynchronizedForAllFollowersId { get; set; }
public DateTime LastSync { get; set; }
public int FetchingErrorCount { get; set; } //TODO: update DAL
}
}

View file

@ -1,4 +1,5 @@
using System;
using System.Diagnostics;
using System.Threading.Tasks;
using System.Xml;
using BirdsiteLive.DAL.Postgres.DataAccessLayers;
@ -47,6 +48,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
Assert.AreEqual(acct, result.Acct);
Assert.AreEqual(lastTweetId, result.LastTweetPostedId);
Assert.AreEqual(lastTweetId, result.LastTweetSynchronizedForAllFollowersId);
Assert.AreEqual(0, result.FetchingErrorCount);
Assert.IsTrue(result.Id > 0);
}
@ -83,13 +85,47 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
var updatedLastTweetId = 1600L;
var updatedLastSyncId = 1550L;
var now = DateTime.Now;
await dal.UpdateTwitterUserAsync(result.Id, updatedLastTweetId, updatedLastSyncId, now);
var errors = 15;
await dal.UpdateTwitterUserAsync(result.Id, updatedLastTweetId, updatedLastSyncId, errors, now);
result = await dal.GetTwitterUserAsync(acct);
Assert.AreEqual(acct, result.Acct);
Assert.AreEqual(updatedLastTweetId, result.LastTweetPostedId);
Assert.AreEqual(updatedLastSyncId, result.LastTweetSynchronizedForAllFollowersId);
Assert.AreEqual(errors, result.FetchingErrorCount);
Assert.IsTrue(Math.Abs((now.ToUniversalTime() - result.LastSync).Milliseconds) < 100);
}
[TestMethod]
public async Task CreateUpdate2AndGetUser()
{
var acct = "myid";
var lastTweetId = 1548L;
var dal = new TwitterUserPostgresDal(_settings);
await dal.CreateTwitterUserAsync(acct, lastTweetId);
var result = await dal.GetTwitterUserAsync(acct);
var updatedLastTweetId = 1600L;
var updatedLastSyncId = 1550L;
var now = DateTime.Now;
var errors = 15;
result.LastTweetPostedId = updatedLastTweetId;
result.LastTweetSynchronizedForAllFollowersId = updatedLastSyncId;
result.FetchingErrorCount = errors;
result.LastSync = now;
await dal.UpdateTwitterUserAsync(result);
result = await dal.GetTwitterUserAsync(acct);
Assert.AreEqual(acct, result.Acct);
Assert.AreEqual(updatedLastTweetId, result.LastTweetPostedId);
Assert.AreEqual(updatedLastSyncId, result.LastTweetSynchronizedForAllFollowersId);
Assert.AreEqual(errors, result.FetchingErrorCount);
Assert.IsTrue(Math.Abs((now.ToUniversalTime() - result.LastSync).Milliseconds) < 100);
}
@ -98,7 +134,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
public async Task Update_NoId()
{
var dal = new TwitterUserPostgresDal(_settings);
await dal.UpdateTwitterUserAsync(default, default, default, DateTime.UtcNow);
await dal.UpdateTwitterUserAsync(default, default, default, default, DateTime.UtcNow);
}
[TestMethod]
@ -106,7 +142,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
public async Task Update_NoLastTweetPostedId()
{
var dal = new TwitterUserPostgresDal(_settings);
await dal.UpdateTwitterUserAsync(12, default, default, DateTime.UtcNow);
await dal.UpdateTwitterUserAsync(12, default, default, default, DateTime.UtcNow);
}
[TestMethod]
@ -114,7 +150,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
public async Task Update_NoLastTweetSynchronizedForAllFollowersId()
{
var dal = new TwitterUserPostgresDal(_settings);
await dal.UpdateTwitterUserAsync(12, 9556, default, DateTime.UtcNow);
await dal.UpdateTwitterUserAsync(12, 9556, default, default, DateTime.UtcNow);
}
[TestMethod]
@ -122,7 +158,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
public async Task Update_NoLastSync()
{
var dal = new TwitterUserPostgresDal(_settings);
await dal.UpdateTwitterUserAsync(12, 9556, 65, default);
await dal.UpdateTwitterUserAsync(12, 9556, 65, default, default);
}
[TestMethod]
@ -216,7 +252,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
{
var user = allUsers[i];
var date = i % 2 == 0 ? oldest : newest;
await dal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, date);
await dal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, 0, date);
}
var result = await dal.GetAllTwitterUsersAsync(10);

View file

@ -0,0 +1,307 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Moderation.Actions;
using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Pipeline.Processors;
using BirdsiteLive.Twitter;
using BirdsiteLive.Twitter.Models;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;
namespace BirdsiteLive.Pipeline.Tests.Processors
{
[TestClass]
public class RefreshTwitterUserStatusProcessorTests
{
[TestMethod]
public async Task ProcessAsync_Test()
{
#region Stubs
var userId1 = 1;
var userId2 = 2;
var users = new List<SyncTwitterUser>
{
new SyncTwitterUser
{
Id = userId1
},
new SyncTwitterUser
{
Id = userId2
}
};
#endregion
#region Mocks
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.IsAny<string>()))
.Returns(new TwitterUser
{
Protected = false
});
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
#endregion
var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object);
var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
#region Validations
Assert.AreEqual(2 , result.Length);
Assert.IsTrue(result.Any(x => x.User.Id == userId1));
Assert.IsTrue(result.Any(x => x.User.Id == userId2));
twitterUserServiceMock.VerifyAll();
twitterUserDalMock.VerifyAll();
removeTwitterAccountActionMock.VerifyAll();
#endregion
}
[TestMethod]
public async Task ProcessAsync_ResetErrorCount_Test()
{
#region Stubs
var userId1 = 1;
var users = new List<SyncTwitterUser>
{
new SyncTwitterUser
{
Id = userId1,
FetchingErrorCount = 100
}
};
#endregion
#region Mocks
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.IsAny<string>()))
.Returns(new TwitterUser
{
Protected = false
});
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
#endregion
var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object);
var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
#region Validations
Assert.AreEqual(1, result.Length);
Assert.IsTrue(result.Any(x => x.User.Id == userId1));
Assert.AreEqual(0, result.First().User.FetchingErrorCount);
twitterUserServiceMock.VerifyAll();
twitterUserDalMock.VerifyAll();
removeTwitterAccountActionMock.VerifyAll();
#endregion
}
[TestMethod]
public async Task ProcessAsync_Unfound_Test()
{
#region Stubs
var userId1 = 1;
var acct1 = "user1";
var userId2 = 2;
var acct2 = "user2";
var users = new List<SyncTwitterUser>
{
new SyncTwitterUser
{
Id = userId1,
Acct = acct1
},
new SyncTwitterUser
{
Id = userId2,
Acct = acct2
}
};
#endregion
#region Mocks
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == acct1)))
.Returns(new TwitterUser
{
Protected = false
});
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == acct2)))
.Returns((TwitterUser) null);
twitterUserServiceMock
.Setup(x => x.PurgeUser(It.Is<string>(y => y == acct2)));
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
twitterUserDalMock
.Setup(x => x.GetTwitterUserAsync(It.Is<string>(y => y == acct2)))
.ReturnsAsync(new SyncTwitterUser
{
Id = userId2,
FetchingErrorCount = 0
});
twitterUserDalMock
.Setup(x => x.UpdateTwitterUserAsync(It.Is<SyncTwitterUser>(y => y.Id == userId2 && y.FetchingErrorCount == 1)))
.Returns(Task.CompletedTask);
var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
#endregion
var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object);
var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
#region Validations
Assert.AreEqual(1, result.Length);
Assert.IsTrue(result.Any(x => x.User.Id == userId1));
twitterUserServiceMock.VerifyAll();
twitterUserDalMock.VerifyAll();
removeTwitterAccountActionMock.VerifyAll();
#endregion
}
[TestMethod]
public async Task ProcessAsync_Unfound_OverThreshold_Test()
{
#region Stubs
var userId1 = 1;
var acct1 = "user1";
var userId2 = 2;
var acct2 = "user2";
var users = new List<SyncTwitterUser>
{
new SyncTwitterUser
{
Id = userId1,
Acct = acct1
},
new SyncTwitterUser
{
Id = userId2,
Acct = acct2
}
};
#endregion
#region Mocks
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == acct1)))
.Returns(new TwitterUser
{
Protected = false
});
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == acct2)))
.Returns((TwitterUser)null);
twitterUserServiceMock
.Setup(x => x.PurgeUser(It.Is<string>(y => y == acct2)));
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
twitterUserDalMock
.Setup(x => x.GetTwitterUserAsync(It.Is<string>(y => y == acct2)))
.ReturnsAsync(new SyncTwitterUser
{
Id = userId2,
FetchingErrorCount = 500
});
var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
removeTwitterAccountActionMock
.Setup(x => x.ProcessAsync(It.Is<SyncTwitterUser>(y => y.Id == userId2)))
.Returns(Task.CompletedTask);
#endregion
var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object);
var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
#region Validations
Assert.AreEqual(1, result.Length);
Assert.IsTrue(result.Any(x => x.User.Id == userId1));
twitterUserServiceMock.VerifyAll();
twitterUserDalMock.VerifyAll();
removeTwitterAccountActionMock.VerifyAll();
#endregion
}
[TestMethod]
public async Task ProcessAsync_Protected_Test()
{
#region Stubs
var userId1 = 1;
var acct1 = "user1";
var userId2 = 2;
var acct2 = "user2";
var users = new List<SyncTwitterUser>
{
new SyncTwitterUser
{
Id = userId1,
Acct = acct1
},
new SyncTwitterUser
{
Id = userId2,
Acct = acct2
}
};
#endregion
#region Mocks
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == acct1)))
.Returns(new TwitterUser
{
Protected = false
});
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == acct2)))
.Returns(new TwitterUser
{
Protected = true
});
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
#endregion
var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object);
var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
#region Validations
Assert.AreEqual(1, result.Length);
Assert.IsTrue(result.Any(x => x.User.Id == userId1));
twitterUserServiceMock.VerifyAll();
twitterUserDalMock.VerifyAll();
removeTwitterAccountActionMock.VerifyAll();
#endregion
}
}
}

View file

@ -21,16 +21,16 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var userId1 = 1;
var userId2 = 2;
var users = new List<UserWithTweetsToSync>
var users = new List<UserWithDataToSync>
{
new UserWithTweetsToSync
new UserWithDataToSync
{
User = new SyncTwitterUser
{
Id = userId1
}
},
new UserWithTweetsToSync
new UserWithDataToSync
{
User = new SyncTwitterUser
{

View file

@ -4,6 +4,7 @@ using System.Threading;
using System.Threading.Tasks;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Pipeline.Processors;
using BirdsiteLive.Twitter;
using BirdsiteLive.Twitter.Models;
@ -27,9 +28,14 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
LastTweetPostedId = -1
};
var user1WtData = new UserWithDataToSync
{
User = user1,
};
var users = new[]
{
user1
user1WtData
};
var tweets = new[]
@ -57,14 +63,12 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
It.Is<int>(y => y == user1.Id),
It.Is<long>(y => y == tweets.Last().Id),
It.Is<long>(y => y == tweets.Last().Id),
It.Is<int>(y => y == 0),
It.IsAny<DateTime>()
))
.Returns(Task.CompletedTask);
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == user1.Acct)))
.Returns(new TwitterUser {Protected = false});
var logger = new Mock<ILogger<RetrieveTweetsProcessor>>(MockBehavior.Strict);
#endregion
@ -94,9 +98,14 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
LastTweetSynchronizedForAllFollowersId = 46
};
var user1WtData = new UserWithDataToSync
{
User = user1,
};
var users = new[]
{
user1
user1WtData
};
var tweets = new[]
@ -129,9 +138,6 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == user1.Acct)))
.Returns(new TwitterUser { Protected = false });
var logger = new Mock<ILogger<RetrieveTweetsProcessor>>(MockBehavior.Strict);
#endregion
@ -147,7 +153,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
Assert.AreEqual(users.Length, usersResult.Length);
Assert.AreEqual(users[0].Acct, usersResult[0].User.Acct);
Assert.AreEqual(users[0].User.Acct, usersResult[0].User.Acct);
Assert.AreEqual(tweets.Length, usersResult[0].Tweets.Length);
#endregion
}
@ -164,9 +170,14 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
LastTweetSynchronizedForAllFollowersId = 46
};
var user1WtData = new UserWithDataToSync
{
User = user1,
};
var users = new[]
{
user1
user1WtData
};
var tweets = new[]
@ -199,9 +210,6 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUser(It.Is<string>(y => y == user1.Acct)))
.Returns(new TwitterUser { Protected = false });
var logger = new Mock<ILogger<RetrieveTweetsProcessor>>(MockBehavior.Strict);
#endregion
@ -216,7 +224,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
logger.VerifyAll();
Assert.AreEqual(users.Length, usersResult.Length);
Assert.AreEqual(users[0].Acct, usersResult[0].User.Acct);
Assert.AreEqual(users[0].User.Acct, usersResult[0].User.Acct);
Assert.AreEqual(tweets.Length, usersResult[0].Tweets.Length);
#endregion
}

View file

@ -41,7 +41,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
}
};
var usersWithTweets = new UserWithTweetsToSync
var usersWithTweets = new UserWithDataToSync
{
Tweets = new []
{
@ -65,6 +65,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
It.Is<int>(y => y == user.Id),
It.Is<long>(y => y == tweet2.Id),
It.Is<long>(y => y == tweet2.Id),
It.Is<int>(y => y == 0),
It.IsAny<DateTime>()
))
.Returns(Task.CompletedTask);
@ -107,7 +108,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
}
};
var usersWithTweets = new UserWithTweetsToSync
var usersWithTweets = new UserWithDataToSync
{
Tweets = new[]
{
@ -130,6 +131,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
It.Is<int>(y => y == user.Id),
It.Is<long>(y => y == tweet3.Id),
It.Is<long>(y => y == tweet2.Id),
It.Is<int>(y => y == 0),
It.IsAny<DateTime>()
))
.Returns(Task.CompletedTask);
@ -181,7 +183,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
}
};
var usersWithTweets = new UserWithTweetsToSync
var usersWithTweets = new UserWithDataToSync
{
Tweets = new[]
{
@ -205,6 +207,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
It.Is<int>(y => y == user.Id),
It.Is<long>(y => y == tweet3.Id),
It.Is<long>(y => y == tweet2.Id),
It.Is<int>(y => y == 0),
It.IsAny<DateTime>()
))
.Returns(Task.CompletedTask);

View file

@ -26,7 +26,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var userId2 = 3;
var userAcct = "user";
var userWithTweets = new UserWithTweetsToSync()
var userWithTweets = new UserWithDataToSync()
{
Tweets = new []
{
@ -93,7 +93,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var userId2 = 3;
var userAcct = "user";
var userWithTweets = new UserWithTweetsToSync()
var userWithTweets = new UserWithDataToSync()
{
Tweets = new[]
{
@ -163,7 +163,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var userId2 = 3;
var userAcct = "user";
var userWithTweets = new UserWithTweetsToSync()
var userWithTweets = new UserWithDataToSync()
{
Tweets = new[]
{
@ -237,7 +237,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var userId2 = 3;
var userAcct = "user";
var userWithTweets = new UserWithTweetsToSync()
var userWithTweets = new UserWithDataToSync()
{
Tweets = new[]
{
@ -306,7 +306,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var userId2 = 3;
var userAcct = "user";
var userWithTweets = new UserWithTweetsToSync()
var userWithTweets = new UserWithDataToSync()
{
Tweets = new[]
{
@ -375,7 +375,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
var userId2 = 3;
var userAcct = "user";
var userWithTweets = new UserWithTweetsToSync()
var userWithTweets = new UserWithDataToSync()
{
Tweets = new[]
{

View file

@ -27,6 +27,7 @@ namespace BirdsiteLive.Pipeline.Tests
It.IsAny<CancellationToken>()))
.Returns(Task.Delay(0));
var refreshTwitterUserStatusProcessor = new Mock<IRefreshTwitterUserStatusProcessor>(MockBehavior.Strict);
var retrieveTweetsProcessor = new Mock<IRetrieveTweetsProcessor>(MockBehavior.Strict);
var retrieveFollowersProcessor = new Mock<IRetrieveFollowersProcessor>(MockBehavior.Strict);
var sendTweetsToFollowersProcessor = new Mock<ISendTweetsToFollowersProcessor>(MockBehavior.Strict);
@ -34,7 +35,7 @@ namespace BirdsiteLive.Pipeline.Tests
var logger = new Mock<ILogger<StatusPublicationPipeline>>();
#endregion
var pipeline = new StatusPublicationPipeline(retrieveTweetsProcessor.Object, retrieveTwitterUsersProcessor.Object, retrieveFollowersProcessor.Object, sendTweetsToFollowersProcessor.Object, saveProgressionProcessor.Object, logger.Object);
var pipeline = new StatusPublicationPipeline(retrieveTweetsProcessor.Object, retrieveTwitterUsersProcessor.Object, retrieveFollowersProcessor.Object, sendTweetsToFollowersProcessor.Object, saveProgressionProcessor.Object, refreshTwitterUserStatusProcessor.Object, logger.Object);
await pipeline.ExecuteAsync(ct.Token);
#region Validations