mz_clusterd_test_driver/
text.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! The text script format: a hand-writable, `datadriven`-style command file.
11//!
12//! A script is a sequence of stanzas, each a command and its expected output:
13//!
14//! ```text
15//! write-single-ts shard=data ts=0 count=5000
16//! ----
17//! wrote 5000
18//!
19//! count id=1001 ts=5
20//! ----
21//! 10000
22//! ```
23//!
24//! A stanza is a command (a directive line plus an optional indentation-structured
25//! body) up to a `----` separator, then the expected output up to a blank line. The
26//! `----` block is the assertion; `REWRITE=1` regenerates it (see [`crate::script`]).
27//! A `#` at column 0 is a comment; an indented `#0` is a column reference in MIR.
28//! Comments and blank lines are preserved across a rewrite.
29//!
30//! Command bodies are indentation-structured: `define-schema`/`write-rows`/`peek`
31//! carry rows or columns, and `define` carries `import`/`build`/`export`
32//! sub-commands, with a `build`'s MIR as its own deeper-indented sub-body.
33
34use std::collections::BTreeMap;
35
36use anyhow::{Context, anyhow, bail, ensure};
37
38use crate::script::{BuildSpec, ColumnSpec, Command, ConfigSetting, ExportSpec, ImportSpec};
39
40/// One element of a parsed script file, retained so a `REWRITE` reproduces the
41/// file faithfully.
42pub enum Item {
43    /// A blank line or column-0 `#` comment, kept verbatim.
44    Verbatim(String),
45    /// A command and its expected output.
46    Stanza(Stanza),
47}
48
49/// A command stanza: the input block, the expected output, and the parsed command.
50pub struct Stanza {
51    /// The directive line plus body, verbatim (for rewrite).
52    pub input: String,
53    /// The expected output block (no trailing newline).
54    pub expected: String,
55    /// The parsed command.
56    pub command: Command,
57}
58
59/// Parse a script file into items: each command stanza plus the comments and blank
60/// lines around it.
61pub fn parse_file(content: &str) -> anyhow::Result<Vec<Item>> {
62    let lines: Vec<&str> = content.lines().collect();
63    let mut items = Vec::new();
64    let mut i = 0;
65    while i < lines.len() {
66        let line = lines[i];
67        // Blank lines and column-0 comments are preserved as-is.
68        if line.trim().is_empty() || line.starts_with('#') {
69            items.push(Item::Verbatim(line.to_string()));
70            i += 1;
71            continue;
72        }
73        // A stanza: slurp the input block up to the `----` separator.
74        let start = i;
75        while i < lines.len() && lines[i] != "----" {
76            i += 1;
77        }
78        ensure!(
79            i < lines.len(),
80            "stanza starting at line {} has no `----` separator",
81            start + 1
82        );
83        let input = lines[start..i].join("\n");
84        i += 1; // consume `----`
85        // The expected output runs to the next blank line (or end of file).
86        let exp_start = i;
87        while i < lines.len() && !lines[i].trim().is_empty() {
88            i += 1;
89        }
90        let expected = lines[exp_start..i].join("\n");
91        let command = parse_command(&input)
92            .with_context(|| format!("parsing stanza at line {}", start + 1))?;
93        items.push(Item::Stanza(Stanza {
94            input,
95            expected,
96            command,
97        }));
98    }
99    Ok(items)
100}
101
102/// Reproduce a script file with each stanza's expected output replaced by its
103/// actual output, for `REWRITE`. `actuals` has one entry per [`Item::Stanza`], in
104/// order.
105pub fn rewrite(items: &[Item], actuals: &[String]) -> String {
106    let mut out = String::new();
107    let mut next = 0;
108    for item in items {
109        match item {
110            Item::Verbatim(line) => {
111                out.push_str(line);
112                out.push('\n');
113            }
114            Item::Stanza(stanza) => {
115                let actual = &actuals[next];
116                next += 1;
117                out.push_str(&stanza.input);
118                out.push_str("\n----\n");
119                out.push_str(actual);
120                if !actual.is_empty() {
121                    out.push('\n');
122                }
123            }
124        }
125    }
126    out
127}
128
129/// A non-blank line of a command block, with its leading-space indentation.
130struct Line {
131    indent: usize,
132    text: String,
133}
134
135/// Split a command block into non-blank lines, recording each line's indentation.
136fn lex(block: &str) -> Vec<Line> {
137    block
138        .lines()
139        .filter_map(|raw| {
140            let trimmed = raw.trim_end();
141            if trimmed.trim().is_empty() {
142                return None;
143            }
144            let indent = trimmed.len() - trimmed.trim_start().len();
145            Some(Line {
146                indent,
147                text: trimmed.trim_start().to_string(),
148            })
149        })
150        .collect()
151}
152
153/// Split body lines into `(header, body)` groups: each header is a line at the
154/// minimum indentation, owning the following more-indented lines as its body.
155fn group(lines: &[Line]) -> anyhow::Result<Vec<(&Line, &[Line])>> {
156    if lines.is_empty() {
157        return Ok(vec![]);
158    }
159    let base = lines.iter().map(|l| l.indent).min().expect("non-empty");
160    let mut groups = Vec::new();
161    let mut i = 0;
162    while i < lines.len() {
163        ensure!(
164            lines[i].indent == base,
165            "inconsistent indentation: `{}`",
166            lines[i].text
167        );
168        let start = i + 1;
169        let mut j = start;
170        while j < lines.len() && lines[j].indent > base {
171            j += 1;
172        }
173        groups.push((&lines[i], &lines[start..j]));
174        i = j;
175    }
176    Ok(groups)
177}
178
179/// Render body lines as text, dedented to the body's minimum indentation so their
180/// relative structure (which MIR depends on) is preserved.
181fn body_text(lines: &[Line]) -> String {
182    if lines.is_empty() {
183        return String::new();
184    }
185    let base = lines.iter().map(|l| l.indent).min().expect("non-empty");
186    lines
187        .iter()
188        .map(|l| format!("{}{}", " ".repeat(l.indent - base), l.text))
189        .collect::<Vec<_>>()
190        .join("\n")
191}
192
193/// Split a header line into whitespace-separated tokens, keeping `"..."` strings
194/// and `[...]` lists intact.
195fn tokenize(s: &str) -> anyhow::Result<Vec<String>> {
196    let mut tokens = Vec::new();
197    let mut cur = String::new();
198    let mut depth = 0i32;
199    let mut in_quote = false;
200    for c in s.chars() {
201        match c {
202            '"' => {
203                in_quote = !in_quote;
204                cur.push(c);
205            }
206            '[' if !in_quote => {
207                depth += 1;
208                cur.push(c);
209            }
210            ']' if !in_quote => {
211                depth -= 1;
212                cur.push(c);
213            }
214            c if c.is_whitespace() && !in_quote && depth == 0 => {
215                if !cur.is_empty() {
216                    tokens.push(std::mem::take(&mut cur));
217                }
218            }
219            c => cur.push(c),
220        }
221    }
222    ensure!(!in_quote, "unterminated string in `{s}`");
223    ensure!(depth == 0, "unterminated list in `{s}`");
224    if !cur.is_empty() {
225        tokens.push(cur);
226    }
227    Ok(tokens)
228}
229
230/// Parse a header line into its verb, `key=value` arguments, and bare flags.
231fn parse_header(header: &str) -> anyhow::Result<(String, BTreeMap<String, String>, Vec<String>)> {
232    let mut tokens = tokenize(header)?.into_iter();
233    let verb = tokens.next().context("empty command")?;
234    let mut args = BTreeMap::new();
235    let mut flags = Vec::new();
236    for token in tokens {
237        match token.split_once('=') {
238            Some((key, value)) => {
239                args.insert(key.to_string(), value.to_string());
240            }
241            None => flags.push(token),
242        }
243    }
244    Ok((verb, args, flags))
245}
246
247fn req<'a>(args: &'a BTreeMap<String, String>, key: &str) -> anyhow::Result<&'a str> {
248    args.get(key)
249        .map(String::as_str)
250        .ok_or_else(|| anyhow!("missing argument `{key}`"))
251}
252
253fn req_u64(args: &BTreeMap<String, String>, key: &str) -> anyhow::Result<u64> {
254    req(args, key)?
255        .parse()
256        .with_context(|| format!("argument `{key}` is not an integer"))
257}
258
259fn opt_u64(args: &BTreeMap<String, String>, key: &str) -> anyhow::Result<Option<u64>> {
260    args.get(key)
261        .map(|v| {
262            v.parse()
263                .with_context(|| format!("argument `{key}` is not an integer"))
264        })
265        .transpose()
266}
267
268fn opt_usize(args: &BTreeMap<String, String>, key: &str) -> anyhow::Result<Option<usize>> {
269    args.get(key)
270        .map(|v| {
271            v.parse()
272                .with_context(|| format!("argument `{key}` is not an integer"))
273        })
274        .transpose()
275}
276
277fn opt_string(args: &BTreeMap<String, String>, key: &str) -> Option<String> {
278    args.get(key).cloned()
279}
280
281/// Parse a `[a,b,c]` list of `usize`s (`[]` is empty).
282fn parse_usize_list(s: &str) -> anyhow::Result<Vec<usize>> {
283    let inner = s
284        .strip_prefix('[')
285        .and_then(|s| s.strip_suffix(']'))
286        .ok_or_else(|| anyhow!("expected a list like `[0,1]`, got `{s}`"))?;
287    if inner.trim().is_empty() {
288        return Ok(vec![]);
289    }
290    inner
291        .split(',')
292        .map(|part| {
293            part.trim()
294                .parse()
295                .with_context(|| format!("bad list element `{part}`"))
296        })
297        .collect()
298}
299
300/// Parse an `export` sub-command into an [`ExportSpec`]. The `kind=` argument
301/// selects the variant (defaulting to `index`); each kind takes its own arguments.
302fn parse_export(args: &BTreeMap<String, String>) -> anyhow::Result<ExportSpec> {
303    let on_id = req_u64(args, "on")?;
304    Ok(
305        match args.get("kind").map(String::as_str).unwrap_or("index") {
306            "index" => ExportSpec::Index {
307                index_id: req_u64(args, "index")?,
308                on_id,
309                key: parse_usize_list(req(args, "key")?)?,
310            },
311            "materialized-view" => ExportSpec::MaterializedView {
312                sink_id: req_u64(args, "sink")?,
313                on_id,
314                shard: req(args, "shard")?.to_string(),
315                schema: opt_string(args, "schema"),
316            },
317            "subscribe" => ExportSpec::Subscribe {
318                sink_id: req_u64(args, "sink")?,
319                on_id,
320                schema: opt_string(args, "schema"),
321                up_to: opt_u64(args, "up-to")?,
322            },
323            "copy-to" => bail!("copy-to export is not implemented"),
324            other => bail!("unknown export kind `{other}`"),
325        },
326    )
327}
328
329/// Parse body lines as rows of raw space-separated value tokens (quotes intact).
330/// The tokens are typed against the schema server-side; see `cell_from_token`.
331fn rows_from_body(body: &[Line]) -> anyhow::Result<Vec<Vec<String>>> {
332    body.iter().map(|l| tokenize(&l.text)).collect()
333}
334
335/// Parse body lines as dyncfg settings: `name type value`.
336fn settings_from_body(body: &[Line]) -> anyhow::Result<Vec<ConfigSetting>> {
337    body.iter()
338        .map(|l| {
339            let tokens = tokenize(&l.text)?;
340            ensure!(
341                tokens.len() == 3,
342                "config setting needs `name type value`, got `{}`",
343                l.text
344            );
345            Ok(ConfigSetting {
346                name: tokens[0].clone(),
347                ty: tokens[1].clone(),
348                value: tokens[2].clone(),
349            })
350        })
351        .collect()
352}
353
354/// Parse body lines as column declarations: `name type [nullable]`.
355fn columns_from_body(body: &[Line]) -> anyhow::Result<Vec<ColumnSpec>> {
356    body.iter()
357        .map(|l| {
358            let tokens = tokenize(&l.text)?;
359            ensure!(
360                tokens.len() >= 2,
361                "column needs `name type [nullable]`, got `{}`",
362                l.text
363            );
364            Ok(ColumnSpec {
365                name: tokens[0].clone(),
366                ty: tokens[1].clone(),
367                nullable: tokens.get(2).is_some_and(|t| t == "nullable"),
368            })
369        })
370        .collect()
371}
372
373/// Parse a `create-dataflow` body of `import`/`build`/`export` sub-commands. The
374/// directive's bare flags carry the dataflow-level options (`optimize`).
375fn parse_create_dataflow(
376    args: &BTreeMap<String, String>,
377    flags: &[String],
378    body: &[Line],
379) -> anyhow::Result<Command> {
380    let name = opt_string(args, "name");
381    let as_of = req_u64(args, "as-of")?;
382    let optimize = flags.iter().any(|f| f == "optimize");
383    let mut imports = Vec::new();
384    let mut builds = Vec::new();
385    let mut exports = Vec::new();
386    for (header, sub_body) in group(body)? {
387        let (verb, args, _flags) = parse_header(&header.text)?;
388        match verb.as_str() {
389            "import" => {
390                if let Some(index_id) = args.get("index") {
391                    imports.push(ImportSpec::Index {
392                        index_id: index_id
393                            .parse()
394                            .with_context(|| format!("bad index id `{index_id}`"))?,
395                    });
396                } else {
397                    imports.push(ImportSpec::Source {
398                        id: req_u64(&args, "source")?,
399                        shard: req(&args, "shard")?.to_string(),
400                        schema: opt_string(&args, "schema"),
401                        upper: req_u64(&args, "upper")?,
402                    });
403                }
404            }
405            "build" => {
406                ensure!(!sub_body.is_empty(), "`build` needs a MIR body");
407                builds.push(BuildSpec {
408                    id: req_u64(&args, "id")?,
409                    expr: body_text(sub_body),
410                });
411            }
412            "export" => exports.push(parse_export(&args)?),
413            other => bail!("unknown `create-dataflow` sub-command `{other}`"),
414        }
415    }
416    Ok(Command::CreateDataflow {
417        name,
418        imports,
419        builds,
420        exports,
421        as_of,
422        optimize,
423    })
424}
425
426/// Parse one command block (directive line plus indentation-structured body).
427fn parse_command(input: &str) -> anyhow::Result<Command> {
428    let lines = lex(input);
429    let (header, body) = lines.split_first().context("empty command")?;
430    ensure!(header.indent == 0, "directive must not be indented");
431    let (verb, args, flags) = parse_header(&header.text)?;
432    let command = match verb.as_str() {
433        "define-schema" => Command::DefineSchema {
434            name: req(&args, "name")?.to_string(),
435            columns: columns_from_body(body)?,
436        },
437        "write-single-ts" => Command::WriteSingleTs {
438            shard: req(&args, "shard")?.to_string(),
439            schema: opt_string(&args, "schema"),
440            ts: req_u64(&args, "ts")?,
441            count: req_u64(&args, "count")?,
442            start: opt_u64(&args, "start")?.unwrap_or(0),
443            row_bytes: opt_usize(&args, "row-bytes")?,
444        },
445        "write-spread" => Command::WriteSpread {
446            shard: req(&args, "shard")?.to_string(),
447            schema: opt_string(&args, "schema"),
448            count: req_u64(&args, "count")?,
449            n_ts: req_u64(&args, "n-ts")?,
450            start: opt_u64(&args, "start")?.unwrap_or(0),
451            row_bytes: opt_usize(&args, "row-bytes")?,
452        },
453        "write-rows" => Command::WriteRows {
454            shard: req(&args, "shard")?.to_string(),
455            schema: opt_string(&args, "schema"),
456            ts: req_u64(&args, "ts")?,
457            rows: rows_from_body(body)?,
458        },
459        "define-index" => Command::DefineIndex {
460            source_id: req_u64(&args, "source")?,
461            index_id: req_u64(&args, "index")?,
462            shard: req(&args, "shard")?.to_string(),
463            schema: opt_string(&args, "schema"),
464            key: parse_usize_list(req(&args, "key")?)?,
465            as_of: req_u64(&args, "as-of")?,
466            upper: req_u64(&args, "upper")?,
467        },
468        "schedule" => Command::Schedule {
469            id: req_u64(&args, "id")?,
470        },
471        "allow-compaction" => Command::AllowCompaction {
472            id: req_u64(&args, "id")?,
473            frontier: req_u64(&args, "frontier")?,
474        },
475        "allow-writes" => Command::AllowWrites {
476            id: req_u64(&args, "id")?,
477        },
478        "await-frontier" => Command::AwaitFrontier {
479            id: req_u64(&args, "id")?,
480            ts: req_u64(&args, "ts")?,
481            timeout_secs: opt_u64(&args, "timeout-secs")?,
482            allow_timeout: flags.iter().any(|f| f == "allow-timeout"),
483        },
484        "count" => Command::Count {
485            id: req_u64(&args, "id")?,
486            ts: req_u64(&args, "ts")?,
487        },
488        "peek" => Command::Peek {
489            id: req_u64(&args, "id")?,
490            schema: opt_string(&args, "schema"),
491            ts: req_u64(&args, "ts")?,
492        },
493        "await-subscribe" => Command::AwaitSubscribe {
494            id: req_u64(&args, "id")?,
495            up_to: req_u64(&args, "up-to")?,
496            timeout_secs: opt_u64(&args, "timeout-secs")?,
497        },
498        "create-dataflow" => parse_create_dataflow(&args, &flags, body)?,
499        "create-instance" => Command::CreateInstance {
500            expiration_offset: opt_string(&args, "expiration-offset"),
501            arrangement_dictionary_compression: args
502                .get("arrangement-dictionary-compression")
503                .map(|v| v.parse())
504                .transpose()
505                .context("argument `arrangement-dictionary-compression` is not a bool")?
506                .unwrap_or(false),
507            // Same `name type value` body rows as `update-configuration`.
508            initial_config: settings_from_body(body)?,
509        },
510        "update-configuration" => Command::UpdateConfiguration {
511            updates: settings_from_body(body)?,
512        },
513        "reconnect" => Command::Reconnect,
514        "initialization-complete" => Command::InitializationComplete,
515        other => bail!("unknown command `{other}`"),
516    };
517    Ok(command)
518}
519
520#[cfg(test)]
521mod tests {
522    use super::*;
523
524    /// A simple command parses its args; absent optionals default.
525    #[mz_ore::test]
526    fn parses_simple_command() {
527        let cmd = parse_command("write-single-ts shard=data ts=0 count=5000").unwrap();
528        assert_eq!(
529            cmd,
530            Command::WriteSingleTs {
531                shard: "data".to_string(),
532                schema: None,
533                ts: 0,
534                count: 5000,
535                start: 0,
536                row_bytes: None,
537            }
538        );
539    }
540
541    /// A flag and a list argument parse.
542    #[mz_ore::test]
543    fn parses_flag_and_list() {
544        let cmd =
545            parse_command("await-frontier id=1001 ts=1 timeout-secs=3 allow-timeout").unwrap();
546        assert_eq!(
547            cmd,
548            Command::AwaitFrontier {
549                id: 1001,
550                ts: 1,
551                timeout_secs: Some(3),
552                allow_timeout: true,
553            }
554        );
555        let cmd =
556            parse_command("define-index source=1000 index=1001 shard=d key=[0] as-of=0 upper=1")
557                .unwrap();
558        assert_eq!(
559            cmd,
560            Command::DefineIndex {
561                source_id: 1000,
562                index_id: 1001,
563                shard: "d".to_string(),
564                schema: None,
565                key: vec![0],
566                as_of: 0,
567                upper: 1,
568            }
569        );
570    }
571
572    /// `define-schema` and `write-rows` parse their indented bodies, typing values.
573    #[mz_ore::test]
574    fn parses_bodies() {
575        let cmd = parse_command(
576            "define-schema name=events\n  key bigint\n  flag boolean\n  label text nullable",
577        )
578        .unwrap();
579        assert_eq!(
580            cmd,
581            Command::DefineSchema {
582                name: "events".to_string(),
583                columns: vec![
584                    ColumnSpec {
585                        name: "key".to_string(),
586                        ty: "bigint".to_string(),
587                        nullable: false
588                    },
589                    ColumnSpec {
590                        name: "flag".to_string(),
591                        ty: "boolean".to_string(),
592                        nullable: false
593                    },
594                    ColumnSpec {
595                        name: "label".to_string(),
596                        ty: "text".to_string(),
597                        nullable: true
598                    },
599                ],
600            }
601        );
602
603        let cmd = parse_command(
604            "write-rows shard=ev schema=events ts=1\n  1000 true alpha\n  1001 false null",
605        )
606        .unwrap();
607        assert_eq!(
608            cmd,
609            Command::WriteRows {
610                shard: "ev".to_string(),
611                schema: Some("events".to_string()),
612                ts: 1,
613                rows: vec![
614                    vec!["1000".to_string(), "true".to_string(), "alpha".to_string()],
615                    vec!["1001".to_string(), "false".to_string(), "null".to_string()],
616                ],
617            }
618        );
619    }
620
621    /// `create-dataflow` parses its sub-commands, keeping a `build`'s MIR (its
622    /// deeper body) with relative indentation preserved; the export kind defaults
623    /// to `index`.
624    #[mz_ore::test]
625    fn parses_create_dataflow_with_mir() {
626        let input = "create-dataflow name=count as-of=0\n  import index=1001\n  build id=2000\n    Reduce aggregates=[count(*)]\n      Get u1000\n  export index=2001 on=2000 key=[0]";
627        let cmd = parse_command(input).unwrap();
628        assert_eq!(
629            cmd,
630            Command::CreateDataflow {
631                name: Some("count".to_string()),
632                imports: vec![ImportSpec::Index { index_id: 1001 }],
633                builds: vec![BuildSpec {
634                    id: 2000,
635                    expr: "Reduce aggregates=[count(*)]\n  Get u1000".to_string(),
636                }],
637                exports: vec![ExportSpec::Index {
638                    index_id: 2001,
639                    on_id: 2000,
640                    key: vec![0]
641                }],
642                as_of: 0,
643                optimize: false,
644            }
645        );
646
647        // The `optimize` flag on the directive line is picked up.
648        let optimized = parse_command(
649            "create-dataflow name=j as-of=0 optimize\n  import source=1000 shard=l upper=1\n  build id=2000\n    Get u1000\n  export index=2001 on=2000 key=[0]",
650        )
651        .unwrap();
652        assert!(matches!(
653            optimized,
654            Command::CreateDataflow { optimize: true, .. }
655        ));
656    }
657
658    /// `create-dataflow` parses the sink export kinds: a materialized-view sink with
659    /// a target shard, and a subscribe sink. The `copy-to` kind is rejected.
660    #[mz_ore::test]
661    fn parses_sink_export_kinds() {
662        let mv = "create-dataflow name=mv as-of=0\n  import source=1000 shard=r upper=1\n  build id=2000\n    Get u1000\n  export kind=materialized-view sink=2001 on=2000 shard=out schema=kv";
663        let Command::CreateDataflow { exports, .. } = parse_command(mv).unwrap() else {
664            panic!("expected create-dataflow");
665        };
666        assert_eq!(
667            exports,
668            vec![ExportSpec::MaterializedView {
669                sink_id: 2001,
670                on_id: 2000,
671                shard: "out".to_string(),
672                schema: Some("kv".to_string()),
673            }]
674        );
675
676        let sub = "create-dataflow name=sub as-of=0\n  import source=1000 shard=s upper=2\n  build id=2000\n    Get u1000\n  export kind=subscribe sink=2001 on=2000 up-to=2";
677        let Command::CreateDataflow { exports, .. } = parse_command(sub).unwrap() else {
678            panic!("expected create-dataflow");
679        };
680        assert_eq!(
681            exports,
682            vec![ExportSpec::Subscribe {
683                sink_id: 2001,
684                on_id: 2000,
685                schema: None,
686                up_to: Some(2),
687            }]
688        );
689
690        // copy-to is named but not implemented.
691        let copy = "create-dataflow name=c as-of=0\n  import source=1000 shard=s upper=1\n  build id=2000\n    Get u1000\n  export kind=copy-to sink=2001 on=2000";
692        assert!(parse_command(copy).is_err());
693    }
694
695    /// `create-instance` parses its optional knobs (defaulting), and
696    /// `update-configuration` parses a `name type value` table (empty when bodyless).
697    #[mz_ore::test]
698    fn parses_handshake_config() {
699        assert_eq!(
700            parse_command("create-instance").unwrap(),
701            Command::CreateInstance {
702                expiration_offset: None,
703                arrangement_dictionary_compression: false,
704                initial_config: vec![],
705            }
706        );
707        assert_eq!(
708            parse_command(
709                "create-instance expiration-offset=30s arrangement-dictionary-compression=true"
710            )
711            .unwrap(),
712            Command::CreateInstance {
713                expiration_offset: Some("30s".to_string()),
714                arrangement_dictionary_compression: true,
715                initial_config: vec![],
716            }
717        );
718        // The create-time snapshot is a `name type value` body, like `update-configuration`.
719        assert_eq!(
720            parse_command("create-instance\n  enable_my_flag bool true").unwrap(),
721            Command::CreateInstance {
722                expiration_offset: None,
723                arrangement_dictionary_compression: false,
724                initial_config: vec![ConfigSetting {
725                    name: "enable_my_flag".to_string(),
726                    ty: "bool".to_string(),
727                    value: "true".to_string(),
728                }],
729            }
730        );
731
732        assert_eq!(
733            parse_command("update-configuration").unwrap(),
734            Command::UpdateConfiguration { updates: vec![] }
735        );
736        assert_eq!(
737            parse_command("update-configuration\n  enable_my_flag bool true\n  my_dur duration 1s")
738                .unwrap(),
739            Command::UpdateConfiguration {
740                updates: vec![
741                    ConfigSetting {
742                        name: "enable_my_flag".to_string(),
743                        ty: "bool".to_string(),
744                        value: "true".to_string(),
745                    },
746                    ConfigSetting {
747                        name: "my_dur".to_string(),
748                        ty: "duration".to_string(),
749                        value: "1s".to_string(),
750                    },
751                ],
752            }
753        );
754    }
755
756    /// `await-subscribe` parses its arguments.
757    #[mz_ore::test]
758    fn parses_await_subscribe() {
759        let cmd = parse_command("await-subscribe id=2001 up-to=2 timeout-secs=5").unwrap();
760        assert_eq!(
761            cmd,
762            Command::AwaitSubscribe {
763                id: 2001,
764                up_to: 2,
765                timeout_secs: Some(5),
766            }
767        );
768    }
769
770    /// A file splits into stanzas, preserving comments and blanks, and round-trips
771    /// through a rewrite when the actual output equals the expected.
772    #[mz_ore::test]
773    fn parses_and_rewrites_file() {
774        let content =
775            "# a comment\nschedule id=1001\n----\nok\n\ncount id=1001 ts=5\n----\n10000\n";
776        let items = parse_file(content).unwrap();
777        let stanzas: Vec<_> = items
778            .iter()
779            .filter_map(|i| match i {
780                Item::Stanza(s) => Some(s),
781                Item::Verbatim(_) => None,
782            })
783            .collect();
784        assert_eq!(stanzas.len(), 2);
785        assert_eq!(stanzas[0].command, Command::Schedule { id: 1001 });
786        assert_eq!(stanzas[0].expected, "ok");
787        assert_eq!(stanzas[1].expected, "10000");
788
789        // Rewriting with the same outputs reproduces the file.
790        let actuals = vec!["ok".to_string(), "10000".to_string()];
791        assert_eq!(rewrite(&items, &actuals), content);
792    }
793}
mz_clusterd_test_driver/text.rs

mz_clusterd_test_driver/
text.rs
