Repository navigation
Expand file tree
/
Copy pathTestRulesAsync.cs
More file actions
121 lines (103 loc) · 5.06 KB
/
Copy pathTestRulesAsync.cs
File metadata and controls
121 lines (103 loc) · 5.06 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
#pragma warning disable CS0618 // allow testing obsolete methods
using StackExchange.Redis;
using NRedisStack.Literals.Enums;
using NRedisStack.DataTypes;
using Xunit;
using NRedisStack.RedisStackCommands;
namespace NRedisStack.Tests.TimeSeries.TestAPI;
public class TestRulesAsync(EndpointsFixture endpointsFixture) : AbstractNRedisStackTest(endpointsFixture)
{
[SkipIfRedisTheory(Is.Enterprise)]
[MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))]
public async Task TestRulesAdditionDeletion(string endpointId)
{
SkipClusterPre8(endpointId);
var key = CreateKeyName();
var db = GetCleanDatabase(endpointId);
var ts = db.TS();
await ts.CreateAsync(key);
var allAggregations = (TsAggregation[])Enum.GetValues(typeof(TsAggregation));
// Filter out CountNan and CountAll on Redis versions < 8.6.0 as they are not supported
var aggregations = EndpointsFixture.RedisVersion >= new Version("8.5.0")
? allAggregations
: allAggregations.Where(a => a != TsAggregation.CountNan && a != TsAggregation.CountAll).ToArray();
foreach (var aggregation in aggregations)
{
await ts.CreateAsync($"{key}:{aggregation}");
}
var timeBucket = 50L;
var rules = new List<TimeSeriesRule>();
var rulesMap = new Dictionary<TsAggregation, TimeSeriesRule>();
foreach (var aggregation in aggregations)
{
var rule = new TimeSeriesRule($"{key}:{aggregation}", timeBucket, aggregation);
rules.Add(rule);
rulesMap[aggregation] = rule;
Assert.True(await ts.CreateRuleAsync(key, rule));
var info = await ts.InfoAsync(key);
Assert.Equal(rules, info.Rules);
}
foreach (var aggregation in aggregations)
{
var rule = rulesMap[aggregation];
rules.Remove(rule);
Assert.True(await ts.DeleteRuleAsync(key, rule.DestKey));
var info = await ts.InfoAsync(key);
Assert.Equal(rules, info.Rules);
}
await db.KeyDeleteAsync(aggregations.Select(i => (RedisKey)$"{key}:{i}").ToArray());
}
[SkipIfRedisTheory(Is.Enterprise)]
[MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))]
public async Task TestNonExistingSrc(string endpointId)
{
SkipClusterPre8(endpointId);
var key = CreateKeyName();
var aggKey = $"{key}:{TsAggregation.Avg}";
var db = GetCleanDatabase(endpointId);
var ts = db.TS();
await ts.CreateAsync(aggKey);
var rule = new TimeSeriesRule(aggKey, 50, TsAggregation.Avg);
var ex = await Assert.ThrowsAsync<RedisServerException>(async () => await ts.CreateRuleAsync(key, rule));
Assert.Equal("ERR TSDB: the key does not exist", ex.Message);
ex = await Assert.ThrowsAsync<RedisServerException>(async () => await ts.DeleteRuleAsync(key, aggKey));
Assert.Equal("ERR TSDB: the key does not exist", ex.Message);
await db.KeyDeleteAsync(aggKey);
}
[SkipIfRedisTheory(Is.Enterprise)]
[MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))]
public async Task TestNonExisitingDestinaion(string endpointId)
{
SkipClusterPre8(endpointId);
var key = CreateKeyName();
var aggKey = $"{key}:{TsAggregation.Avg}";
var db = GetCleanDatabase(endpointId);
var ts = db.TS();
await ts.CreateAsync(key);
var rule = new TimeSeriesRule(aggKey, 50, TsAggregation.Avg);
var ex = await Assert.ThrowsAsync<RedisServerException>(async () => await ts.CreateRuleAsync(key, rule));
Assert.Equal("ERR TSDB: the key does not exist", ex.Message);
ex = await Assert.ThrowsAsync<RedisServerException>(async () => await ts.DeleteRuleAsync(key, aggKey));
Assert.Equal("ERR TSDB: compaction rule does not exist", ex.Message);
}
[SkipIfRedisTheory(Is.Enterprise)]
[MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))]
public async Task TestAlignTimestampAsync(string endpointId)
{
SkipClusterPre8(endpointId);
IDatabase db = GetCleanDatabase(endpointId);
var ts = db.TS();
await ts.CreateAsync("{ts}ts1");
await ts.CreateAsync("{ts}ts2");
await ts.CreateAsync("{ts}ts3");
TimeSeriesRule rule1 = new("{ts}ts2", 10, TsAggregation.Count);
await ts.CreateRuleAsync("{ts}ts1", rule1, 0);
TimeSeriesRule rule2 = new("{ts}ts3", 10, TsAggregation.Count);
await ts.CreateRuleAsync("{ts}ts1", rule2, 1);
await ts.AddAsync("{ts}ts1", 1, 1);
await ts.AddAsync("{ts}ts1", 10, 3);
await ts.AddAsync("{ts}ts1", 21, 7);
Assert.Equal(2, (await ts.RangeAsync("{ts}ts2", "-", "+", aggregation: TsAggregation.Count, timeBucket: 10)).Count);
Assert.Single((await ts.RangeAsync("{ts}ts3", "-", "+", aggregation: TsAggregation.Count, timeBucket: 10)));
}
}