kafru 1.0.4

kafru is a Python Celery-inspired queuing library for Rust, using cron for scheduling and SurrealDB for storing queues, metrics, and schedules.
Documentation 
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
122
123
124
125
126
127
128


#[cfg(test)]
mod test_schedule {
    use crate::tests::test_helper::configure_database_env;
    use crate::schedule::{Schedule, ScheduleData, ScheduleStatus, ScheduleListConditions};
    use crate::cron_schedule::CronSchedule;
    use fake::{
        Fake,
        faker::{
            name::en::Name,
            lorem::en::Sentence
        }
    };
    use std::ops::Range;
    use std::collections::HashMap;
    use serde_json::{Value, Number};
    use chrono::{Days, Utc};
    use std::sync::Arc;
    use crate::database::Db;

    #[tokio::test]
    #[serial_test::serial]
    async fn test_create_update_remove(){
        configure_database_env();
        let db_instance = Db::new(None).await;
        assert!(db_instance.is_ok(),"{:?}",db_instance.err());
        let db: Arc<Db> = Arc::new(db_instance.unwrap());
        let schedule: Schedule = Schedule::new(Some(db.clone())).await;

        // Purge records
        let result = schedule.purge().await;
        assert!(result.is_ok(),"{:?}",result.unwrap_err());

        // Create Record
        let result: Result<ScheduleData, String> = schedule.create(ScheduleData {
            name: Some(Name().fake::<String>()),
            queue: Some("default".to_string()),
            cron_expression: Some(CronSchedule::new().set_second("0".to_string()).set_minute("*/5".to_string())),
            handler:Some("mytesthandler".to_string()),
            parameters: Some(HashMap::from([
                (
                    "schedule1".to_string(),
                    Value::String(Sentence( Range {start: 1, end: 5 }).fake::<String>())
                ),
                (
                    "schedule2".to_string(),
                    Value::Number(Number::from(123))
                )
            ])),
            ..Default::default()
        }).await;
        assert!(result.is_ok(),"{}",result.unwrap_err());
        let previous_record: ScheduleData = result.unwrap();
        assert_eq!(previous_record.status,Some(ScheduleStatus::Disabled));
        
        // Update record
        let result: Result<ScheduleData, String> = schedule.update(previous_record.id.clone().unwrap(),ScheduleData {
            status: Some(ScheduleStatus::Enabled),
            start_schedule:Utc::now().checked_add_days(Days::new(1)),
            until_schedule: Utc::now().checked_add_days(Days::new(3)),
            next_schedule: Utc::now().checked_add_days(Days::new(1)),
            ..previous_record.clone()
        }).await;
        assert!(result.is_ok(),"{}",result.unwrap_err());
        let next_record: ScheduleData = result.unwrap();
        assert_eq!(next_record.status,Some(ScheduleStatus::Enabled));
        assert!(next_record.next_schedule > previous_record.next_schedule,"Next Schedule: {:?}   Previous Schedule: {:?}",next_record.next_schedule,previous_record.next_schedule);

        // Get record
        let result: Result<ScheduleData, String> = schedule.get(next_record.id.clone().unwrap()).await;
        assert!(result.is_ok(),"{}",result.unwrap_err());
        let record: ScheduleData = result.unwrap();
        assert_eq!(next_record.id,record.id);

        // Remove the record
        let result: Result<ScheduleData, String> = schedule.remove(record.id.clone().unwrap()).await;
        assert!(result.is_ok(),"{}",result.unwrap_err());
    }
    
    #[tokio::test]
    #[serial_test::serial]
    pub async fn test_list_purge(){
        configure_database_env();
        let db_instance = Db::new(None).await;
        assert!(db_instance.is_ok(),"{:?}",db_instance.err());
        let db: Arc<Db> = Arc::new(db_instance.unwrap());
        let schedule: Schedule = Schedule::new(Some(db.clone())).await;
        
        // Purge records
        let result = schedule.purge().await;
        assert!(result.is_ok(),"{:?}",result.unwrap_err());

        // List schedules
        for i in 1..11 {
            let result: Result<ScheduleData, String> = schedule.create(ScheduleData {
                name: Some(format!("{} - {}",i,Name().fake::<String>())),
                queue: Some("default".to_string()),
                cron_expression: Some(CronSchedule::new().set_minute(format!("*/{}",i))),
                handler:Some("mytesthandler".to_string()),
                until_schedule: Utc::now().checked_add_days(Days::new(3)),
                parameters: Some(HashMap::from([

                    (
                        "schedule1".to_string(),
                        Value::String(Sentence( Range {start: 1, end: 5 }).fake::<String>())
                    ),
                    (
                        "schedule2".to_string(),
                        Value::Number(Number::from(123))
                    )
                ])),
                ..Default::default()
            }).await;
            assert!(result.is_ok(),"{}",result.unwrap_err());
        }
        let result: Result<Vec<ScheduleData>, String> = schedule.list(ScheduleListConditions {
            upcoming: Some(false),
            ..Default::default()
        }).await;
        assert!(result.is_ok(),"{}",result.unwrap_err());
        assert!(result.unwrap().len()>=10);
        
        //// Cleanup using Purge
        let result: Result<u64, String> = schedule.purge().await;
        assert!(result.is_ok(),"{}",result.unwrap_err());
    }
    
}