Improve stream validation; add tests (#108)

.circleci/config.yml

@@ -13,8 +13,8 @@ jobs:
           keys:
             - air01-{{ checksum "Cargo.lock" }}
       - run: |
-          rustup toolchain install nightly-2021-03-24-x86_64-unknown-linux-gnu
-          rustup default nightly-2021-03-24-x86_64-unknown-linux-gnu
+          rustup toolchain install nightly-2021-05-16-x86_64-unknown-linux-gnu
+          rustup default nightly-2021-05-16-x86_64-unknown-linux-gnu
 
           rustup target add wasm32-wasi
           rustup component add rustfmt

air/src/execution/utils/resolve.rs

@@ -80,25 +80,31 @@ fn prepare_variable<'i>(
     variable: &Variable<'_>,
     ctx: &ExecutionCtx<'i>,
 ) -> ExecutionResult<(JValue, Vec<SecurityTetraplet>)> {
-    let resolved = match variable {
-        Variable::Scalar(name) => resolve_to_jvaluable(name, ctx)?,
-        Variable::Stream(name) => {
-            // return an empty stream for not found stream
-            // here it ignores the join behaviour
-            if ctx.data_cache.get(*name).is_none() {
-                Box::new(())
-            } else {
-                resolve_to_jvaluable(name, ctx)?
-            }
-        }
-    };
-
+    let resolved = resolve_variable(variable, ctx)?;
     let tetraplets = resolved.as_tetraplets();
     let jvalue = resolved.into_jvalue();
 
     Ok((jvalue, tetraplets))
 }
 
+fn resolve_variable<'ctx, 'i>(
+    variable: &Variable<'_>,
+    ctx: &'ctx ExecutionCtx<'i>,
+) -> ExecutionResult<Box<dyn JValuable + 'ctx>> {
+    match variable {
+        Variable::Scalar(name) => resolve_to_jvaluable(name, ctx),
+        Variable::Stream(name) => {
+            // return an empty stream for not found stream
+            // here it ignores the join behaviour
+            if ctx.data_cache.get(*name).is_none() {
+                Ok(Box::new(()))
+            } else {
+                resolve_to_jvaluable(name, ctx)
+            }
+        }
+    }
+}
+
 fn prepare_json_path<'i>(
     variable: &Variable<'_>,
     json_path: &str,

air/tests/test_module/integration/join_behaviour.rs

@@ -18,6 +18,7 @@ use air_test_utils::call_vm;
 use air_test_utils::create_avm;
 use air_test_utils::set_variables_call_service;
 use air_test_utils::unit_call_service;
+use air_test_utils::ExecutionTrace;
 
 use serde_json::json;
 

crates/air-parser/src/parser/lexer/call_variable_parser.rs

@@ -23,6 +23,8 @@ use std::convert::TryInto;
 use std::iter::Peekable;
 use std::str::CharIndices;
 
+const STREAM_START_TAG: char = '$';
+
 pub(super) fn try_parse_call_variable(
     string_to_parse: &str,
     start_pos: usize,
@@ -320,8 +322,6 @@ impl<'input> CallVariableParser<'input> {
     }
 }
 
-const STREAM_START_TAG: char = '$';
-
 fn to_variable_and_path(str: &str, pos: usize, should_flatten: bool) -> (&str, &str) {
     let json_path = if should_flatten {
         // -1 to not include the flattening symbol ! to the resulted json path

crates/air-parser/src/parser/lexer/mod.rs

@@ -21,7 +21,7 @@ mod token;
 mod utils;
 
 #[cfg(test)]
-pub mod tests;
+mod tests;
 
 pub use air_lexer::AIRLexer;
 pub use errors::LexerError;

crates/air-parser/src/parser/tests.rs

@@ -163,7 +163,7 @@ fn parse_undefined_variable() {
     let mut validator = super::VariableValidator::new();
     parser
         .parse(source_code, &mut errors, &mut validator, lexer)
-        .expect("parser shoudn't fail");
+        .expect("parser shouldn't fail");
 
     let errors = validator.finalize();
 
@@ -200,7 +200,7 @@ fn parse_undefined_iterable() {
     let mut validator = super::VariableValidator::new();
     parser
         .parse(source_code, &mut errors, &mut validator, lexer)
-        .expect("parser shoudn't fail");
+        .expect("parser shouldn't fail");
 
     let errors = validator.finalize();
 
@@ -215,6 +215,53 @@ fn parse_undefined_iterable() {
     assert!(matches!(parser_error, ParserError::UndefinedIterable(..)));
 }
 
+#[test]
+fn parse_undefined_stream_without_json_path() {
+    let source_code = r#"
+        (call "" "" [$stream])
+        "#;
+
+    let lexer = crate::AIRLexer::new(source_code);
+
+    let parser = crate::AIRParser::new();
+    let mut errors = Vec::new();
+    let mut validator = super::VariableValidator::new();
+    parser
+        .parse(source_code, &mut errors, &mut validator, lexer)
+        .expect("parser shouldn't fail");
+
+    let errors = validator.finalize();
+
+    assert!(errors.is_empty());
+}
+
+#[test]
+fn parse_undefined_stream_with_json_path() {
+    let source_code = r#"
+        (call "" "" [$stream.$json_path])
+        "#;
+
+    let lexer = crate::AIRLexer::new(source_code);
+
+    let parser = crate::AIRParser::new();
+    let mut errors = Vec::new();
+    let mut validator = super::VariableValidator::new();
+    parser
+        .parse(source_code, &mut errors, &mut validator, lexer)
+        .expect("parser shouldn't fail");
+
+    let errors = validator.finalize();
+
+    assert_eq!(errors.len(), 1);
+    let error = &errors[0].error;
+    let parser_error = match error {
+        ParseError::User { error } => error,
+        _ => panic!("unexpected error type"),
+    };
+
+    assert!(matches!(parser_error, ParserError::UndefinedVariable(..)));
+}
+
 #[test]
 fn parse_json_path_complex() {
     use ast::CallOutputValue::*;

crates/air-parser/src/parser/validator.rs

@@ -87,7 +87,7 @@ impl<'i> VariableValidator<'i> {
     pub(super) fn finalize(&self) -> Vec<ErrorRecovery<usize, Token<'i>, ParserError>> {
         let mut errors = Vec::new();
         for (name, span) in self.unresolved_variables.iter() {
-            if !self.contains_variable(name, *span) && !name.starts_with('$') {
+            if !self.contains_variable(name, *span) {
                 add_to_errors(*name, &mut errors, *span, Token::Call);
             }
         }
@@ -138,7 +138,12 @@ impl<'i> VariableValidator<'i> {
     fn met_instr_arg_value(&mut self, instr_arg_value: &CallInstrArgValue<'i>, span: Span) {
         match instr_arg_value {
             CallInstrArgValue::JsonPath { variable, .. } => self.met_variable(variable, span),
-            CallInstrArgValue::Variable(variable) => self.met_variable(variable, span),
+            CallInstrArgValue::Variable(variable) => {
+                // skipping streams here allows treating non-defined streams as empty arrays
+                if let Variable::Scalar(_) = variable {
+                    self.met_variable(variable, span)
+                }
+            }
             _ => {}
         }
     }