slices.network
Auto-indexing service and GraphQL API for AT Protocol Records
quickslice.slices.network/
atproto
gleam
graphql
chadtmiller.com
1import database/repositories/lexicons
2import gleam/json
3import gleam/list
4import gleam/option
5import gleeunit/should
6import lexicon_graphql
7import lexicon_graphql/schema/database
8import swell/schema
9import test_helpers
10
11// Helper to create a status lexicon JSON
12fn create_status_lexicon() -> String {
13 json.object([
14 #("lexicon", json.int(1)),
15 #("id", json.string("xyz.statusphere.status")),
16 #(
17 "defs",
18 json.object([
19 #(
20 "main",
21 json.object([
22 #("type", json.string("record")),
23 #("key", json.string("tid")),
24 #(
25 "record",
26 json.object([
27 #("type", json.string("object")),
28 #(
29 "required",
30 json.array(
31 [json.string("status"), json.string("createdAt")],
32 of: fn(x) { x },
33 ),
34 ),
35 #(
36 "properties",
37 json.object([
38 #(
39 "status",
40 json.object([
41 #("type", json.string("string")),
42 #("maxLength", json.int(300)),
43 ]),
44 ),
45 #(
46 "createdAt",
47 json.object([
48 #("type", json.string("string")),
49 #("format", json.string("datetime")),
50 ]),
51 ),
52 ]),
53 ),
54 ]),
55 ),
56 ]),
57 ),
58 ]),
59 ),
60 ])
61 |> json.to_string
62}
63
64pub fn test_create_mutation_generates_correct_schema_test() {
65 // Setup: Create in-memory database with test lexicons
66 let assert Ok(exec) = test_helpers.create_test_db()
67 let assert Ok(_) = test_helpers.create_lexicon_table(exec)
68
69 // Insert xyz.statusphere.status lexicon
70 let status_lexicon = create_status_lexicon()
71
72 let assert Ok(_) =
73 lexicons.insert(exec, "xyz.statusphere.status", status_lexicon)
74
75 // Get lexicons and parse them
76 let assert Ok(lexicon_records) = lexicons.get_all(exec)
77 let parsed_lexicons =
78 lexicon_records
79 |> list.filter_map(fn(lex) { lexicon_graphql.parse_lexicon(lex.json) })
80
81 // Build schema with empty fetcher
82 let empty_fetcher = fn(_collection, _params) {
83 Ok(#([], option.None, False, False, option.None))
84 }
85
86 let assert Ok(built_schema) =
87 database.build_schema_with_fetcher(
88 parsed_lexicons,
89 empty_fetcher,
90 option.None,
91 option.None,
92 option.None,
93 option.None,
94 option.None,
95 option.None,
96 )
97
98 // Test: Verify that the schema has mutation type
99 let mutation_type_option = schema.get_mutation_type(built_schema)
100
101 // Assert that mutation type exists
102 mutation_type_option
103 |> should.be_some()
104
105 // Verify mutation type has expected fields
106 let assert option.Some(mutation_type) = mutation_type_option
107 let mutation_fields = schema.get_fields(mutation_type)
108
109 // Check that we have 3 mutations per record type (create, update, delete)
110 // For xyz.statusphere.status, we expect:
111 // - createXyzStatusphereStatus
112 // - updateXyzStatusphereStatus
113 // - deleteXyzStatusphereStatus
114 let field_names =
115 mutation_fields
116 |> list.map(schema.field_name)
117
118 field_names
119 |> should.equal([
120 "createXyzStatusphereStatus",
121 "updateXyzStatusphereStatus",
122 "deleteXyzStatusphereStatus",
123 ])
124}
125
126pub fn test_create_mutation_field_signature_test() {
127 // Setup
128 let assert Ok(exec) = test_helpers.create_test_db()
129 let assert Ok(_) = test_helpers.create_lexicon_table(exec)
130 let status_lexicon = create_status_lexicon()
131 let assert Ok(_) =
132 lexicons.insert(exec, "xyz.statusphere.status", status_lexicon)
133 let assert Ok(lexicon_records) = lexicons.get_all(exec)
134 let parsed_lexicons =
135 lexicon_records
136 |> list.filter_map(fn(lex) { lexicon_graphql.parse_lexicon(lex.json) })
137
138 let empty_fetcher = fn(_collection, _params) {
139 Ok(#([], option.None, False, False, option.None))
140 }
141 let assert Ok(built_schema) =
142 database.build_schema_with_fetcher(
143 parsed_lexicons,
144 empty_fetcher,
145 option.None,
146 option.None,
147 option.None,
148 option.None,
149 option.None,
150 option.None,
151 )
152
153 // Get mutation type and find createXyzStatusphereStatus field
154 let assert option.Some(mutation_type) = schema.get_mutation_type(built_schema)
155 let assert option.Some(create_field) =
156 schema.get_field(mutation_type, "createXyzStatusphereStatus")
157
158 // Verify arguments
159 let arguments = schema.field_arguments(create_field)
160 let arg_names = list.map(arguments, schema.argument_name)
161
162 // Should have 'input' and 'rkey' arguments
163 arg_names
164 |> should.equal(["input", "rkey"])
165
166 // Verify input argument is non-null
167 let assert Ok(input_arg) =
168 list.find(arguments, fn(arg) { schema.argument_name(arg) == "input" })
169 let input_type = schema.argument_type(input_arg)
170
171 schema.is_non_null(input_type)
172 |> should.be_true()
173
174 // Verify rkey argument is nullable (optional)
175 let assert Ok(rkey_arg) =
176 list.find(arguments, fn(arg) { schema.argument_name(arg) == "rkey" })
177 let rkey_type = schema.argument_type(rkey_arg)
178
179 schema.is_non_null(rkey_type)
180 |> should.be_false()
181}
182
183pub fn test_update_mutation_field_signature_test() {
184 // Setup
185 let assert Ok(exec) = test_helpers.create_test_db()
186 let assert Ok(_) = test_helpers.create_lexicon_table(exec)
187 let status_lexicon = create_status_lexicon()
188 let assert Ok(_) =
189 lexicons.insert(exec, "xyz.statusphere.status", status_lexicon)
190 let assert Ok(lexicon_records) = lexicons.get_all(exec)
191 let parsed_lexicons =
192 lexicon_records
193 |> list.filter_map(fn(lex) { lexicon_graphql.parse_lexicon(lex.json) })
194
195 let empty_fetcher = fn(_collection, _params) {
196 Ok(#([], option.None, False, False, option.None))
197 }
198 let assert Ok(built_schema) =
199 database.build_schema_with_fetcher(
200 parsed_lexicons,
201 empty_fetcher,
202 option.None,
203 option.None,
204 option.None,
205 option.None,
206 option.None,
207 option.None,
208 )
209
210 // Get mutation type and find updateXyzStatusphereStatus field
211 let assert option.Some(mutation_type) = schema.get_mutation_type(built_schema)
212 let assert option.Some(update_field) =
213 schema.get_field(mutation_type, "updateXyzStatusphereStatus")
214
215 // Verify arguments
216 let arguments = schema.field_arguments(update_field)
217 let arg_names = list.map(arguments, schema.argument_name)
218
219 // Should have 'rkey' and 'input' arguments
220 arg_names
221 |> should.equal(["rkey", "input"])
222
223 // Both should be non-null (required)
224 let assert Ok(rkey_arg) =
225 list.find(arguments, fn(arg) { schema.argument_name(arg) == "rkey" })
226 let rkey_type = schema.argument_type(rkey_arg)
227
228 schema.is_non_null(rkey_type)
229 |> should.be_true()
230
231 let assert Ok(input_arg) =
232 list.find(arguments, fn(arg) { schema.argument_name(arg) == "input" })
233 let input_type = schema.argument_type(input_arg)
234
235 schema.is_non_null(input_type)
236 |> should.be_true()
237}
238
239pub fn test_delete_mutation_field_signature_test() {
240 // Setup
241 let assert Ok(exec) = test_helpers.create_test_db()
242 let assert Ok(_) = test_helpers.create_lexicon_table(exec)
243 let status_lexicon = create_status_lexicon()
244 let assert Ok(_) =
245 lexicons.insert(exec, "xyz.statusphere.status", status_lexicon)
246 let assert Ok(lexicon_records) = lexicons.get_all(exec)
247 let parsed_lexicons =
248 lexicon_records
249 |> list.filter_map(fn(lex) { lexicon_graphql.parse_lexicon(lex.json) })
250
251 let empty_fetcher = fn(_collection, _params) {
252 Ok(#([], option.None, False, False, option.None))
253 }
254 let assert Ok(built_schema) =
255 database.build_schema_with_fetcher(
256 parsed_lexicons,
257 empty_fetcher,
258 option.None,
259 option.None,
260 option.None,
261 option.None,
262 option.None,
263 option.None,
264 )
265
266 // Get mutation type and find deleteXyzStatusphereStatus field
267 let assert option.Some(mutation_type) = schema.get_mutation_type(built_schema)
268 let assert option.Some(delete_field) =
269 schema.get_field(mutation_type, "deleteXyzStatusphereStatus")
270
271 // Verify arguments
272 let arguments = schema.field_arguments(delete_field)
273 let arg_names = list.map(arguments, schema.argument_name)
274
275 // Should have only 'rkey' argument
276 arg_names
277 |> should.equal(["rkey"])
278
279 // Should be non-null (required)
280 let assert Ok(rkey_arg) =
281 list.find(arguments, fn(arg) { schema.argument_name(arg) == "rkey" })
282 let rkey_type = schema.argument_type(rkey_arg)
283
284 schema.is_non_null(rkey_type)
285 |> should.be_true()
286}