From 31a585f13a002261e1f6837a90b95279a8750b38 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Thu, 22 Oct 2026 22:39:36 -0700 Subject: [PATCH] Property-test master tasks against generated answers Runs a READ (user or startup integrity) or one of ten non-READ tasks against a generated stream of responses and noise on the paused-clock mock I/O, and checks every step against a model written from the rules: the requests and confirms the master writes with their sequence numbers, the callbacks and measurements it delivers, how and when the task ends, and the sequence number it continues from. The READ model covers multi-fragment responses with FIR/FIN/CON combinations, wrong sequence numbers, IIN2 request errors and timeouts between fragments. --- dnp3/src/master/tests/mod.rs | 1 + dnp3/src/master/tests/outcomes.rs | 32 +- dnp3/src/master/tests/task_model.rs | 1475 +++++++++++++++++++++++++++ 3 files changed, 1492 insertions(+), 16 deletions(-) create mode 100644 dnp3/src/master/tests/task_model.rs diff --git a/dnp3/src/master/tests/mod.rs b/dnp3/src/master/tests/mod.rs index fd35e8c72..424ea1085 100644 --- a/dnp3/src/master/tests/mod.rs +++ b/dnp3/src/master/tests/mod.rs @@ -18,6 +18,7 @@ mod restart; mod scheduling; mod shared_link; mod startup; +mod task_model; mod time_sync; mod unsolicited; diff --git a/dnp3/src/master/tests/outcomes.rs b/dnp3/src/master/tests/outcomes.rs index 14ff1d5f5..68564adfb 100644 --- a/dnp3/src/master/tests/outcomes.rs +++ b/dnp3/src/master/tests/outcomes.rs @@ -51,9 +51,9 @@ use crate::master::{ use crate::util::session::{RunError, StopReason}; use crate::util::test_support::watchdog; -const OUTSTATION: EndpointAddress = EndpointAddress::raw(1024); -const OTHER: EndpointAddress = EndpointAddress::raw(2048); -const UNKNOWN: EndpointAddress = EndpointAddress::raw(3000); +pub(super) const OUTSTATION: EndpointAddress = EndpointAddress::raw(1024); +pub(super) const OTHER: EndpointAddress = EndpointAddress::raw(2048); +pub(super) const UNKNOWN: EndpointAddress = EndpointAddress::raw(3000); const HANDLE: u32 = 0xDEADCAFE; @@ -64,7 +64,7 @@ async fn barrier(harness: &mut TestHarness) { harness.master.get_decode_level().await.unwrap(); } -struct FixedTime; +pub(super) struct FixedTime; impl AssociationHandler for FixedTime { fn get_current_time(&self) -> Option { @@ -85,7 +85,7 @@ fn echo(request: &[u8]) -> Vec { /// The kinds of task that wait for a response, as requested through the API #[derive(Copy, Clone, Debug)] -enum Kind { +pub(super) enum Kind { Read, DirectOperate, SelectBeforeOperate, @@ -119,7 +119,7 @@ const KINDS: [Kind; 13] = [ /// What the API call returned #[derive(Debug, PartialEq)] -enum Reply { +pub(super) enum Reply { Read(Result<(), TaskError>), Command(Result<(), CommandError>), Restart(Result), @@ -135,7 +135,7 @@ enum Reply { } impl Kind { - fn spawn(self, mut association: AssociationHandle) -> JoinHandle { + pub(super) fn spawn(self, mut association: AssociationHandle) -> JoinHandle { tokio::spawn(async move { let association = &mut association; match self { @@ -220,7 +220,7 @@ impl Kind { }) } - fn task_type(self) -> TaskType { + pub(super) fn task_type(self) -> TaskType { match self { Kind::Read => TaskType::UserRead, Kind::DirectOperate | Kind::SelectBeforeOperate => TaskType::Command, @@ -238,7 +238,7 @@ impl Kind { } /// The function code of the first request - fn function(self) -> FunctionCode { + pub(super) fn function(self) -> FunctionCode { match self { Kind::Read => FunctionCode::Read, Kind::DirectOperate => FunctionCode::DirectOperate, @@ -258,7 +258,7 @@ impl Kind { matches!(self, Kind::Read) } - fn ok(self) -> Reply { + pub(super) fn ok(self) -> Reply { match self { Kind::Read => Reply::Read(Ok(())), Kind::DirectOperate | Kind::SelectBeforeOperate => Reply::Command(Ok(())), @@ -277,7 +277,7 @@ impl Kind { } } - fn failed(self, err: TaskError) -> Reply { + pub(super) fn failed(self, err: TaskError) -> Reply { match self { Kind::Read => Reply::Read(Err(err)), Kind::DirectOperate | Kind::SelectBeforeOperate => { @@ -307,7 +307,7 @@ impl Kind { } /// The successful response to the first request - fn first_response(self, request: &[u8]) -> Vec { + pub(super) fn first_response(self, request: &[u8]) -> Vec { match self { Kind::Read | Kind::EmptyResponse | Kind::DeadBands | Kind::TimeSync => { empty_response(Sequence::new(0)) @@ -687,7 +687,7 @@ async fn every_single_header_task_fails_on_a_response_with_two_headers() { /// Fragments that aren't valid responses: one too short for the IIN, and a solicited response with /// the UNS bit -const MALFORMED: [&[u8]; 2] = [&[0xC0, 0x81, 0x00], &[0xD0, 0x81, 0x00, 0x00]]; +pub(super) const MALFORMED: [&[u8]; 2] = [&[0xC0, 0x81, 0x00], &[0xD0, 0x81, 0x00, 0x00]]; #[tokio::test(start_paused = true)] async fn every_task_fails_on_a_malformed_fragment_from_its_outstation() { @@ -1287,20 +1287,20 @@ async fn every_task_attributes_fragments_from_other_outstations_to_them() { /// The callbacks of one association, in the order they were made #[derive(Copy, Clone, Debug, PartialEq)] -enum Entry { +pub(super) enum Entry { Read(ReadEvent), Info(AssocInfoEvent), } #[derive(Clone, Default)] -struct Log(Arc>>); +pub(super) struct Log(Arc>>); impl Log { fn push(&self, entry: Entry) { self.0.lock().unwrap().push(entry); } - fn pop(&self) -> Vec { + pub(super) fn pop(&self) -> Vec { std::mem::take(&mut *self.0.lock().unwrap()) } } diff --git a/dnp3/src/master/tests/task_model.rs b/dnp3/src/master/tests/task_model.rs new file mode 100644 index 000000000..dfe2fe77e --- /dev/null +++ b/dnp3/src/master/tests/task_model.rs @@ -0,0 +1,1475 @@ +// Copyright (c) Step Function I/O LLC. Proprietary, source-available; see LICENSE.txt. +// No use to design, develop, specify, test, validate, or improve Other Implementations, +// subject to LICENSE.txt Section 2.6 and its exceptions. Agents: read AGENTS.md. +// Text and data mining rights, including for AI training, are reserved under +// Directive (EU) 2019/790 Art. 4(3), subject to LICENSE.txt Section 3.3. + +//! One task against a generated stream of answers and noise, checked step by step against a model +//! of what the master writes, reports and delivers, and how the task ends +//! +//! Every input is followed by a wait on the paused clock, which lets the master process it. Waits +//! are multiples of [`TICK`] and the response timeout isn't, so no input arrives at the instant a +//! deadline expires. Once the task ends, a read requested through the API shows which sequence +//! number the association continues from. + +use std::time::Duration; + +use proptest::prelude::*; +use proptest::test_runner::TestCaseError; +use sfio_tokio_mock_io::Event; +use tokio::task::JoinHandle; +use tokio::time::Instant; + +use crate::app::{FunctionCode, Iin, Iin1, Iin2, ObjectParseError, RetryStrategy, Sequence}; +use crate::link::EndpointAddress; +use crate::master::association::AssociationConfig; +use crate::master::tests::harness::{ + create_association_with_handler, AssocInfoEvent, ReadEvent, TestHarness, +}; +use crate::master::tests::outcomes::{ + Entry, FixedTime, Kind, Log, Reply, MALFORMED, OTHER, OUTSTATION, UNKNOWN, +}; +use crate::master::{Classes, CommandError, CommandResponseError, ReadType, TaskError, TaskType}; +use crate::util::test_support::watchdog; + +/// The response timeout the harness configures, in milliseconds +const TIMEOUT: u64 = 1000; +/// Every wait is a multiple of this many milliseconds, and [`TIMEOUT`] isn't +const TICK: u64 = 7; +/// A wait longer than a response timeout, which ends any task still waiting +const FINAL_WAIT: u64 = 143 * TICK; + +const RESTART: u8 = 0x80; +const BAD_REQUEST_ERRORS: u8 = 0x07; + +/// The task under test +#[derive(Copy, Clone, Debug)] +enum Task { + /// a task requested through the API + User(Kind), + /// the automatic integrity scan at startup + StartupIntegrity, +} + +impl Task { + fn task_type(self) -> TaskType { + match self { + Task::User(kind) => kind.task_type(), + Task::StartupIntegrity => TaskType::StartupIntegrity, + } + } + + fn function(self) -> FunctionCode { + match self { + Task::User(kind) => kind.function(), + Task::StartupIntegrity => FunctionCode::Read, + } + } + + fn is_read(self) -> bool { + matches!(self, Task::User(Kind::Read) | Task::StartupIntegrity) + } + + fn read_type(self) -> ReadType { + match self { + Task::StartupIntegrity => ReadType::StartupIntegrity, + Task::User(_) => ReadType::SinglePoll, + } + } + + /// A read the user asked for fails if the outstation reported a request error and returned no + /// values; the startup integrity scan completes anyway + fn fails_without_data(self) -> bool { + matches!(self, Task::User(_)) + } + + /// Unsolicited responses with data are accepted only after the startup integrity scan + fn accepts_unsolicited_data(self) -> bool { + matches!(self, Task::User(_)) + } + + /// True if a correct response is a control echo, and a wrong one fails as a command + fn is_command(self) -> bool { + matches!( + self, + Task::User(Kind::DirectOperate | Kind::SelectBeforeOperate) + ) + } +} + +#[derive(Copy, Clone, Debug, PartialEq)] +enum Source { + Outstation, + /// another outstation, with its own association + Other, + /// an address without an association + Unknown, +} + +impl Source { + fn address(self) -> EndpointAddress { + match self { + Source::Outstation => OUTSTATION, + Source::Other => OTHER, + Source::Unknown => UNKNOWN, + } + } + + fn index(self) -> usize { + match self { + Source::Outstation => 0, + Source::Other => 1, + Source::Unknown => 2, + } + } +} + +/// The objects of a fragment. Reads and unsolicited responses use the measurement kinds; other +/// tasks use `Correct`, `Unexpected` and `Malformed`. +#[derive(Copy, Clone, Debug, PartialEq)] +enum Objects { + None, + /// one analog input event (g32v2) + AnalogEvent, + /// one static analog input (g30v2) + AnalogStatic, + /// a common time of occurrence, then a binary input event with relative time + CtoRelative, + /// a binary input event with relative time and no common time of occurrence before it + RelativeWithoutCto, + /// an object header cut short after its group + Malformed, + /// the objects the task expects + Correct, + /// objects that parse, but that the task doesn't expect + Unexpected, +} + +/// What a read or unsolicited response delivers, and how the handler sees it +#[derive(Copy, Clone, Debug, PartialEq)] +enum Measurement { + Analog, + Binary, +} + +/// What the master can make of a read or unsolicited fragment's objects +#[derive(Copy, Clone, Debug, PartialEq)] +enum Content { + Valid(Option), + Malformed, + RelativeWithoutCto, +} + +impl Objects { + fn content(self) -> Content { + match self { + Objects::None | Objects::Correct | Objects::Unexpected => Content::Valid(None), + Objects::AnalogEvent | Objects::AnalogStatic => { + Content::Valid(Some(Measurement::Analog)) + } + Objects::CtoRelative => Content::Valid(Some(Measurement::Binary)), + Objects::RelativeWithoutCto => Content::RelativeWithoutCto, + Objects::Malformed => Content::Malformed, + } + } + + fn measurement_bytes(self) -> Vec { + match self { + Objects::None | Objects::Correct | Objects::Unexpected => vec![], + Objects::AnalogEvent => vec![32, 2, 0x17, 1, 0, 0x01, 7, 0], + Objects::AnalogStatic => vec![30, 2, 0x00, 0, 0, 0x01, 7, 0], + Objects::CtoRelative => vec![ + 51, 1, 0x07, 1, 0, 0, 0, 0, 0, 0, 2, 3, 0x17, 1, 0, 0x81, 0x10, 0x00, + ], + Objects::RelativeWithoutCto => vec![2, 3, 0x17, 1, 0, 0x81, 0x10, 0x00], + Objects::Malformed => vec![0x01], + } + } +} + +/// What a non-read task can make of a response's objects +#[derive(Copy, Clone, Debug, PartialEq)] +enum Reaction { + Correct, + Unexpected, + Malformed, +} + +impl Reaction { + fn of(objects: Objects) -> Self { + match objects { + Objects::Correct => Reaction::Correct, + Objects::Malformed => Reaction::Malformed, + Objects::None + | Objects::AnalogEvent + | Objects::AnalogStatic + | Objects::CtoRelative + | Objects::RelativeWithoutCto + | Objects::Unexpected => Reaction::Unexpected, + } + } + + fn bytes(self, kind: Kind, request: &[u8]) -> Vec { + match self { + Reaction::Correct => kind.first_response(request).split_off(4), + // a task that expects objects gets none, and one that expects none gets a time delay + Reaction::Unexpected => match kind { + Kind::EmptyResponse | Kind::DeadBands | Kind::TimeSync => { + vec![52, 1, 0x07, 1, 0x02, 0x00] + } + _ => vec![], + }, + Reaction::Malformed => vec![0x01], + } + } +} + +#[derive(Copy, Clone, Debug)] +enum SeqChoice { + /// the sequence number the master expects + Expected, + /// the one before it, as a repeat of the previous fragment has + Previous, + /// the one after it + Next, +} + +#[derive(Copy, Clone, Debug)] +enum Fir { + /// set on the first fragment of a response only + Proper, + Flipped, +} + +/// A solicited response fragment for the task's current request +#[derive(Copy, Clone, Debug)] +struct Answer { + source: Source, + seq: SeqChoice, + fir: Fir, + fin: bool, + con: bool, + restart: bool, + iin2: u8, + objects: Objects, +} + +#[derive(Copy, Clone, Debug)] +struct Unsolicited { + source: Source, + seq: u8, + con: bool, + restart: bool, + objects: Objects, + /// send the last unsolicited fragment from this source again, if there is one + repeat: bool, +} + +impl Unsolicited { + fn bytes(self) -> Vec { + let iin1 = if self.restart { RESTART } else { 0 }; + fragment( + true, + true, + self.con, + true, + self.seq, + iin1, + 0, + &self.objects.measurement_bytes(), + ) + } +} + +#[derive(Copy, Clone, Debug)] +enum Input { + Answer(Answer), + Unsolicited(Unsolicited), + /// a fragment that isn't a valid response, one of [`MALFORMED`] + Malformed(Source, usize), + /// a request through the API that doesn't involve the outstation + Api, + Silence, +} + +#[derive(Copy, Clone, Debug)] +struct Step { + input: Input, + /// milliseconds to wait after the input + wait: u64, +} + +#[expect(clippy::too_many_arguments, reason = "every field of the header")] +fn fragment( + fir: bool, + fin: bool, + con: bool, + uns: bool, + seq: u8, + iin1: u8, + iin2: u8, + objects: &[u8], +) -> Vec { + let control = (u8::from(fir) << 7) + | (u8::from(fin) << 6) + | (u8::from(con) << 5) + | (u8::from(uns) << 4) + | (seq & 0x0F); + let function = if uns { 0x82 } else { 0x81 }; + let mut bytes = vec![control, function, iin1, iin2]; + bytes.extend_from_slice(objects); + bytes +} + +fn next_seq(seq: u8) -> u8 { + seq.wrapping_add(1) & 0x0F +} + +fn previous_seq(seq: u8) -> u8 { + seq.wrapping_add(15) & 0x0F +} + +/// What the master wrote +#[derive(Clone, Debug, PartialEq)] +enum Write { + /// a request with FIR and FIN, its function code and sequence number + Request(FunctionCode, u8), + /// a solicited confirm, and its sequence number + Confirm(u8), + /// an unsolicited confirm, and its sequence number + UnsolicitedConfirm(u8), + Other(Vec), +} + +impl Write { + fn classify(bytes: &[u8]) -> Self { + if let [control, function, rest @ ..] = bytes { + let seq = control & 0x0F; + match (control & 0xF0, FunctionCode::from(*function)) { + (0xC0, Some(FunctionCode::Confirm)) if rest.is_empty() => { + return Write::Confirm(seq) + } + (0xD0, Some(FunctionCode::Confirm)) if rest.is_empty() => { + return Write::UnsolicitedConfirm(seq) + } + (0xC0, Some(function)) if function != FunctionCode::Confirm => { + return Write::Request(function, seq) + } + _ => {} + } + } + Write::Other(bytes.to_vec()) + } +} + +/// Everything observable after one step +#[derive(Clone, Debug, Default, PartialEq)] +struct StepObs { + writes: Vec, + /// how many fragments the master read + reads: usize, + /// read handler callbacks of the association under test + main: Vec, + /// association information callbacks of the association under test + info: Vec, + /// callbacks of the other outstation's association + other: Vec, +} + +/// How the task ended +#[derive(Copy, Clone, Debug, PartialEq)] +enum End { + /// with the sequence number of the last response + Success(u8), + Failure(TaskError), + /// a command whose response doesn't echo the request + CommandMismatch, +} + +impl End { + fn event(self, task: Task) -> AssocInfoEvent { + match self { + End::Success(seq) => { + AssocInfoEvent::TaskSuccess(task.task_type(), task.function(), Sequence::new(seq)) + } + End::Failure(err) => AssocInfoEvent::TaskFailure(task.task_type(), err), + End::CommandMismatch => { + AssocInfoEvent::TaskFailure(task.task_type(), TaskError::UnexpectedResponseHeaders) + } + } + } + + fn reply(self, kind: Kind) -> Reply { + match self { + End::Success(_) => kind.ok(), + End::Failure(err) => kind.failed(err), + End::CommandMismatch => Reply::Command(Err(CommandError::Response( + CommandResponseError::HeaderCountMismatch, + ))), + } + } +} + +/// How and when the task ended, and what a read requested next through the API then writes +#[derive(Clone, Debug)] +struct Ending { + end: End, + time: u64, + probe: StepObs, +} + +/// One allowed outcome of a step: what's observed, and how the task ended in it, if it did +#[derive(Clone, Debug)] +struct Allowed { + obs: StepObs, + ending: Option, +} + +/// Whether the restart IIN bit of the outstation must be acted on +#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +enum RestartSeen { + /// nothing reported it, or only fragments the master ignores: another outstation's, or a + /// response to another request + Never, + /// only a solicited fragment that breaks the FIR, FIN or CON rules reported it: the master + /// rejects it before looking at its IIN or objects, and either is right + Maybe, + /// a response the master processed reported it, even if it then rejected the response + Must, +} + +enum State { + Read { + first: bool, + seq: u8, + deadline: u64, + /// IIN2 request errors of every fragment + errors: u8, + /// IIN1 of the last fragment + iin1: u8, + values: bool, + }, + Response { + /// the select of a select-before-operate was answered + operating: bool, + seq: u8, + deadline: u64, + }, + Done, +} + +/// What the master should do, written from the rules rather than the code +struct Model { + task: Task, + now: u64, + /// the sequence number of the association's next request + next_seq: u8, + state: State, + restart: RestartSeen, + /// the last unsolicited fragment accepted from the outstation and from the other one + accepted: [Option>; 2], + /// the last unsolicited fragment sent from each source + sent: [Option; 3], + obs: StepObs, + /// the ways the task may have ended in this step, and when + ends: Vec, + end_time: u64, +} + +impl Model { + fn new(task: Task) -> Self { + let state = if task.is_read() { + State::Read { + first: true, + seq: 0, + deadline: TIMEOUT, + errors: 0, + iin1: 0, + values: false, + } + } else { + State::Response { + operating: false, + seq: 0, + deadline: TIMEOUT, + } + }; + let obs = StepObs { + writes: vec![Write::Request(task.function(), 0)], + info: vec![AssocInfoEvent::TaskStart( + task.task_type(), + task.function(), + Sequence::new(0), + )], + ..StepObs::default() + }; + Self { + task, + now: 0, + next_seq: 1, + state, + restart: RestartSeen::Never, + accepted: [None, None], + sent: [None, None, None], + obs, + ends: vec![], + end_time: 0, + } + } + + fn done(&self) -> bool { + matches!(self.state, State::Done) + } + + fn deadline(&self) -> Option { + match self.state { + State::Read { deadline, .. } | State::Response { deadline, .. } => Some(deadline), + State::Done => None, + } + } + + fn saw_restart(&mut self, restart: bool, seen: RestartSeen) { + if restart { + self.restart = self.restart.max(seen); + } + } + + fn end(&mut self, ends: Vec) { + self.state = State::Done; + self.ends = ends; + self.end_time = self.now; + } + + /// How long to wait after an input: the step's wait, but no more than a short while after the + /// task ends, so that nothing the end leads to, such as clearing a restart, times out in the step + fn wait(&self, wait: u64) -> u64 { + match self.deadline() { + None => TICK, + Some(deadline) => wait.min(deadline.saturating_sub(self.now).saturating_add(TICK)), + } + } + + /// If a wait reaches the deadline, the part of it that ends a millisecond before + fn before_deadline(&self, wait: u64) -> Option { + let remaining = self.deadline()?.saturating_sub(self.now); + (remaining <= wait).then(|| remaining.saturating_sub(1)) + } + + /// Wait, ending the task if its deadline passes + fn advance(&mut self, wait: u64) { + let until = self.now.saturating_add(wait); + if let Some(deadline) = self.deadline() { + if deadline <= until { + self.now = deadline; + self.end(vec![End::Failure(TaskError::ResponseTimeout)]); + } + } + self.now = until; + } + + /// The allowed outcomes of the step that just ended + fn take_step(&mut self) -> Vec { + let obs = std::mem::take(&mut self.obs); + let ends = std::mem::take(&mut self.ends); + if ends.is_empty() { + return vec![Allowed { obs, ending: None }]; + } + // the restart is cleared once the task ends + let clear_restart: &[bool] = match self.restart { + RestartSeen::Never => &[false], + RestartSeen::Maybe => &[false, true], + RestartSeen::Must => &[true], + }; + let mut allowed = vec![]; + for end in ends { + for clear in clear_restart { + let mut obs = obs.clone(); + obs.info.push(end.event(self.task)); + // the next request continues from the task's last sequence number + let (task_type, function) = if *clear { + (TaskType::ClearRestartBit, FunctionCode::Write) + } else { + (TaskType::UserRead, FunctionCode::Read) + }; + let next = StepObs { + writes: vec![Write::Request(function, self.next_seq)], + info: vec![AssocInfoEvent::TaskStart( + task_type, + function, + Sequence::new(self.next_seq), + )], + ..StepObs::default() + }; + // clearing the restart starts at once, and the read waits for its response + let probe = if *clear { + obs.writes.extend(next.writes); + obs.info.extend(next.info); + StepObs::default() + } else { + next + }; + allowed.push(Allowed { + obs, + ending: Some(Ending { + end, + time: self.end_time, + probe, + }), + }); + } + } + allowed + } + + /// Apply an input, returning the fragment to send and where it comes from. `request` is the + /// last request the master wrote, which a control response echoes. + fn input(&mut self, input: Input, request: &[u8]) -> Option<(EndpointAddress, Vec)> { + match input { + Input::Silence | Input::Api => None, + Input::Malformed(source, which) => { + self.obs.reads = 1; + if source == Source::Outstation { + self.end(vec![End::Failure(TaskError::Transport)]); + } + let bytes = MALFORMED.get(which).copied().unwrap_or_default(); + Some((source.address(), bytes.to_vec())) + } + Input::Unsolicited(unsolicited) => { + let mut unsolicited = match self.sent[unsolicited.source.index()] { + Some(last) if unsolicited.repeat => last, + _ => unsolicited, + }; + if unsolicited.source != Source::Outstation { + unsolicited.restart = false; + } + self.sent[unsolicited.source.index()] = Some(unsolicited); + self.obs.reads = 1; + let bytes = unsolicited.bytes(); + self.unsolicited(unsolicited, &bytes); + Some((unsolicited.source.address(), bytes)) + } + Input::Answer(answer) => { + self.obs.reads = 1; + let bytes = self.answer(answer, request); + Some((answer.source.address(), bytes)) + } + } + } + + fn unsolicited(&mut self, unsolicited: Unsolicited, bytes: &[u8]) { + let accepted = match unsolicited.source { + Source::Unknown => return, + Source::Outstation => &self.accepted[0], + Source::Other => &self.accepted[1], + }; + let restart = unsolicited.restart; + let seq = Sequence::new(unsolicited.seq); + let content = unsolicited.objects.content(); + let null = unsolicited.objects.measurement_bytes().is_empty(); + let is_duplicate = accepted.as_deref() == Some(bytes); + + // the IIN of an unsolicited response from an association is acted on even if the response + // is a duplicate, comes before the startup integrity scan, or has objects it can't process + self.saw_restart(restart, RestartSeen::Must); + if unsolicited.source == Source::Outstation + && !(self.task.accepts_unsolicited_data() || null) + { + return; + } + if is_duplicate { + self.record_unsolicited(unsolicited.source, AssocInfoEvent::Unsolicited(true, seq)); + if unsolicited.con { + self.obs + .writes + .push(Write::UnsolicitedConfirm(unsolicited.seq)); + } + return; + } + let Content::Valid(measurement) = content else { + // neither processed nor confirmed, and not remembered + return; + }; + match unsolicited.source { + Source::Outstation => { + self.accepted[0] = Some(bytes.to_vec()); + self.obs + .main + .extend(delivery(ReadType::Unsolicited, measurement)); + } + Source::Other => { + self.accepted[1] = Some(bytes.to_vec()); + self.obs.other.extend( + delivery(ReadType::Unsolicited, measurement) + .into_iter() + // the other association's handler only records analog inputs + .filter(|event| *event != ReadEvent::BinaryInput) + .map(Entry::Read), + ); + } + Source::Unknown => {} + } + self.record_unsolicited(unsolicited.source, AssocInfoEvent::Unsolicited(false, seq)); + if unsolicited.con { + self.obs + .writes + .push(Write::UnsolicitedConfirm(unsolicited.seq)); + } + } + + fn record_unsolicited(&mut self, source: Source, event: AssocInfoEvent) { + match source { + Source::Outstation => self.obs.info.push(event), + Source::Other => self.obs.other.push(Entry::Info(event)), + Source::Unknown => {} + } + } + + fn answer(&mut self, answer: Answer, request: &[u8]) -> Vec { + let (expected, proper_fir) = match self.state { + State::Read { first, seq, .. } => (seq, first), + State::Response { seq, .. } => (seq, true), + State::Done => (self.next_seq, true), + }; + let seq = match answer.seq { + SeqChoice::Expected => expected, + SeqChoice::Previous => previous_seq(expected), + SeqChoice::Next => next_seq(expected), + }; + let fir = match answer.fir { + Fir::Proper => proper_fir, + Fir::Flipped => !proper_fir, + }; + let iin1 = if answer.restart { RESTART } else { 0 }; + let objects = match self.task { + Task::User(kind) if !self.task.is_read() => { + Reaction::of(answer.objects).bytes(kind, request) + } + Task::User(_) | Task::StartupIntegrity => answer.objects.measurement_bytes(), + }; + let bytes = fragment( + fir, + answer.fin, + answer.con, + false, + seq, + iin1, + answer.iin2, + &objects, + ); + + // responses from other addresses are no concern of this task + if answer.source != Source::Outstation { + return bytes; + } + // nor are responses to other requests, whose IIN may be stale + if seq != expected { + return bytes; + } + if self.task.is_read() { + self.read_fragment(answer, seq, fir, iin1); + } else { + self.response(answer, seq, fir, iin1); + } + bytes + } + + fn read_fragment(&mut self, answer: Answer, seq: u8, fir: bool, iin1: u8) { + let State::Read { first, .. } = self.state else { + return; + }; + // any rule the fragment breaks may be the one reported + let mut broken = vec![]; + if fir && !first { + broken.push(End::Failure(TaskError::UnexpectedFir)); + } + if !fir && first { + broken.push(End::Failure(TaskError::NeverReceivedFir)); + } + if !answer.fin && !answer.con { + broken.push(End::Failure(TaskError::NonFinWithoutCon)); + } + if !broken.is_empty() { + self.saw_restart(answer.restart, RestartSeen::Maybe); + self.end(broken); + return; + } + + // the IIN of the response is acted on even if its objects are rejected + self.saw_restart(answer.restart, RestartSeen::Must); + let measurement = match answer.objects.content() { + Content::Valid(measurement) => measurement, + Content::Malformed => { + self.end(vec![End::Failure(TaskError::MalformedResponse( + ObjectParseError::InsufficientBytes, + ))]); + return; + } + Content::RelativeWithoutCto => { + self.end(vec![End::Failure(TaskError::UnexpectedResponseHeaders)]); + return; + } + }; + + self.obs + .main + .extend(delivery(self.task.read_type(), measurement)); + if answer.con { + self.obs.writes.push(Write::Confirm(seq)); + } + + let fails_without_data = self.task.fails_without_data(); + let next = self.next_seq; + let now = self.now; + let State::Read { + first, + seq: expected, + deadline, + errors, + iin1: last_iin1, + values, + } = &mut self.state + else { + return; + }; + *errors |= answer.iin2 & BAD_REQUEST_ERRORS; + *last_iin1 = iin1; + *values |= measurement.is_some(); + + if answer.fin { + let end = if *errors == 0 || *values || !fails_without_data { + End::Success(seq) + } else { + End::Failure(TaskError::RejectedByIin2(Iin::new( + Iin1::new(*last_iin1), + Iin2::new(*errors), + ))) + }; + self.end(vec![end]); + } else { + // the next fragment has the next sequence number, and its own response timeout + *first = false; + *expected = next; + *deadline = now.saturating_add(TIMEOUT); + self.next_seq = next_seq(next); + } + } + + fn response(&mut self, answer: Answer, seq: u8, fir: bool, iin1: u8) { + let State::Response { operating, .. } = self.state else { + return; + }; + if !(fir && answer.fin) { + self.saw_restart(answer.restart, RestartSeen::Maybe); + self.end(vec![End::Failure(TaskError::MultiFragmentResponse)]); + return; + } + + let reaction = Reaction::of(answer.objects); + let malformed = reaction == Reaction::Malformed; + // a response whose objects parse is confirmed before anything else, even if it's rejected + if answer.con && !malformed { + self.obs.writes.push(Write::Confirm(seq)); + } + // the IIN of the response is acted on even if the response is rejected + self.saw_restart(answer.restart, RestartSeen::Must); + + let mut ends = vec![]; + let rejected = answer.iin2 & BAD_REQUEST_ERRORS != 0; + if rejected { + ends.push(End::Failure(TaskError::RejectedByIin2(Iin::new( + Iin1::new(iin1), + Iin2::new(answer.iin2), + )))); + } + // a rejected response is reported as such, unless its objects can't be parsed either + if !rejected || malformed { + match reaction { + Reaction::Malformed => ends.push(End::Failure(TaskError::MalformedResponse( + ObjectParseError::InsufficientBytes, + ))), + Reaction::Unexpected if self.task.is_command() => ends.push(End::CommandMismatch), + Reaction::Unexpected => { + ends.push(End::Failure(TaskError::UnexpectedResponseHeaders)) + } + Reaction::Correct => { + let selected = + matches!(self.task, Task::User(Kind::SelectBeforeOperate)) && !operating; + if selected { + // then the operate, with the next sequence number + let seq = self.next_seq; + self.obs + .writes + .push(Write::Request(FunctionCode::Operate, seq)); + self.state = State::Response { + operating: true, + seq, + deadline: self.now.saturating_add(TIMEOUT), + }; + self.next_seq = next_seq(seq); + return; + } + ends.push(End::Success(seq)); + } + } + } + self.end(ends); + } +} + +/// The read handler callbacks for one fragment +fn delivery(read_type: ReadType, measurement: Option) -> Vec { + let mut events = vec![ReadEvent::BeginFragment(read_type)]; + match measurement { + Some(Measurement::Analog) => events.push(ReadEvent::AnalogInput), + Some(Measurement::Binary) => events.push(ReadEvent::BinaryInput), + None => {} + } + events.push(ReadEvent::EndFragment(read_type)); + events +} + +/// What the requester got, and when +#[derive(Debug, PartialEq)] +enum ReplyObs { + /// the task has no requester + None, + /// no reply yet + Pending, + /// the reply, and the milliseconds since the task started + Done(Reply, u64), +} + +/// A case run against the master, with the model's allowed outcomes for every step +struct Run { + task: Task, + allowed: Vec>, + observed: Vec, + reply: ReplyObs, + /// what the master did after a read was requested through the API, once the task ended + probe: StepObs, +} + +impl Run { + fn check(&self) -> Result<(), TestCaseError> { + prop_assert_eq!(self.allowed.len(), self.observed.len()); + let mut ending = None; + for (step, (allowed, observed)) in self.allowed.iter().zip(&self.observed).enumerate() { + let matched = allowed.iter().find(|allowed| allowed.obs == *observed); + let Some(matched) = matched else { + return Err(TestCaseError::fail(format!( + "step {step}: observed {observed:#?}\nallowed {allowed:#?}" + ))); + }; + if let Some(matched) = &matched.ending { + prop_assert!(ending.is_none(), "step {}: the task ended twice", step); + ending = Some(matched); + } + } + let Some(ending) = ending else { + return Err(TestCaseError::fail("the task never ended")); + }; + let expected = match self.task { + Task::User(kind) => ReplyObs::Done(ending.end.reply(kind), ending.time), + Task::StartupIntegrity => ReplyObs::None, + }; + prop_assert_eq!(&self.reply, &expected); + prop_assert_eq!(&self.probe, &ending.probe); + Ok(()) + } +} + +async fn start(task: Task) -> (TestHarness, Log) { + let config = match task { + Task::User(_) => AssociationConfig::quiet(), + Task::StartupIntegrity => AssociationConfig { + startup_integrity_classes: Classes::class0(), + // a failed scan isn't retried while the case runs + auto_tasks_retry_strategy: RetryStrategy::new( + Duration::from_secs(3600), + Duration::from_secs(3600), + ), + ..AssociationConfig::quiet() + }, + }; + let mut harness = create_association_with_handler(config, Box::new(FixedTime)).await; + let other = Log::default(); + harness + .master + .add_association( + OTHER, + AssociationConfig::quiet(), + Box::new(other.clone()), + Box::new(other.clone()), + Box::new(other.clone()), + ) + .await + .unwrap(); + (harness, other) +} + +fn observe(harness: &mut TestHarness, other: &Log, last_request: &mut Vec) -> StepObs { + let mut obs = StepObs::default(); + while let Some(event) = harness.io.pop_event() { + match event { + Event::Write(bytes) => { + let write = Write::classify(&bytes); + if matches!(write, Write::Request(..)) { + *last_request = bytes; + } + obs.writes.push(write); + } + Event::Read => obs.reads = obs.reads.saturating_add(1), + Event::WriteErr | Event::ReadErr => obs.writes.push(Write::Other(vec![])), + } + } + obs.main = harness.read_events.pop(); + obs.info = harness.assoc_events.pop(); + obs.other = other.pop(); + obs +} + +async fn wait(ms: u64) { + tokio::time::sleep(Duration::from_millis(ms)).await; +} + +/// Run the task against the steps, stopping once the model says it ended, and silence after them +async fn execute(task: Task, steps: &[Step]) -> Run { + let (mut harness, other) = start(task).await; + let started = Instant::now(); + let reply: Option> = match task { + Task::User(kind) => { + let association = harness.association.clone(); + Some(tokio::spawn(async move { + let reply = kind.spawn(association).await.unwrap(); + (reply, Instant::now()) + })) + } + Task::StartupIntegrity => None, + }; + + let mut model = Model::new(task); + let mut last_request = vec![]; + let mut allowed = vec![]; + let mut observed = vec![]; + + model.advance(TICK); + wait(TICK).await; + allowed.push(model.take_step()); + observed.push(observe(&mut harness, &other, &mut last_request)); + + let silence = Step { + input: Input::Silence, + wait: FINAL_WAIT, + }; + for step in steps.iter().chain([&silence]) { + if model.done() { + break; + } + if let Input::Api = step.input { + let mut master = harness.master.clone(); + drop(tokio::spawn(async move { master.get_decode_level().await })); + } + if let Some((source, bytes)) = model.input(step.input, &last_request) { + harness.set_source(source); + harness.io.read(&bytes); + } + let mut ms = model.wait(step.wait); + // a wait through the deadline is also observed just before it, so a task that times out + // early is seen, whether or not it has a requester whose reply is timed + if let Some(before) = model.before_deadline(ms) { + model.advance(before); + wait(before).await; + allowed.push(model.take_step()); + observed.push(observe(&mut harness, &other, &mut last_request)); + ms = ms.saturating_sub(before); + } + model.advance(ms); + wait(ms).await; + allowed.push(model.take_step()); + observed.push(observe(&mut harness, &other, &mut last_request)); + } + + // a read requested now is the next request, unless clearing the restart is + drop(Kind::Read.spawn(harness.association.clone())); + wait(TICK).await; + let probe = observe(&mut harness, &other, &mut last_request); + + let reply = match reply { + None => ReplyObs::None, + Some(reply) if !reply.is_finished() => ReplyObs::Pending, + Some(reply) => { + let (reply, at) = reply.await.unwrap(); + let elapsed = at.duration_since(started).as_millis(); + ReplyObs::Done(reply, u64::try_from(elapsed).unwrap()) + } + }; + Run { + task, + allowed, + observed, + reply, + probe, + } +} + +/// Run one case on its own paused runtime, under the watchdog +fn run(task: Task, steps: &[Step]) -> Run { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .start_paused(true) + .build() + .unwrap(); + let mut run = None; + runtime.block_on(watchdog(async { + run = Some(execute(task, steps).await); + })); + run.unwrap() +} + +fn source() -> impl Strategy { + prop_oneof![ + 8 => Just(Source::Outstation), + 1 => Just(Source::Other), + 1 => Just(Source::Unknown), + ] +} + +/// mostly short waits, sometimes a large part of a response timeout, and rarely one just short of +/// it or past it +fn wait_ms() -> impl Strategy { + prop_oneof![ + 16 => Just(1u64), + 4 => Just(2), + 2 => Just(35), + 2 => Just(71), + 1 => Just(142), + 1 => Just(143), + ] + .prop_map(|ticks| ticks.saturating_mul(TICK)) +} + +/// mostly no IIN2 bits, then each request error, and the event buffer overflow that isn't one +fn iin2() -> impl Strategy { + prop_oneof![ + 6 => Just(0u8), + 1 => Just(0x01), + 1 => Just(0x02), + 1 => Just(0x04), + 1 => Just(0x08), + ] +} + +fn measurement_objects() -> impl Strategy { + prop_oneof![ + 4 => Just(Objects::None), + 3 => Just(Objects::AnalogEvent), + 2 => Just(Objects::AnalogStatic), + 1 => Just(Objects::CtoRelative), + 1 => Just(Objects::RelativeWithoutCto), + 1 => Just(Objects::Malformed), + ] +} + +fn response_objects() -> impl Strategy { + prop_oneof![ + 6 => Just(Objects::Correct), + 1 => Just(Objects::Unexpected), + 1 => Just(Objects::Malformed), + ] +} + +fn answer(read: bool) -> impl Strategy { + let seq = prop_oneof![ + 8 => Just(SeqChoice::Expected), + 1 => Just(SeqChoice::Previous), + 1 => Just(SeqChoice::Next), + ]; + let fir = prop_oneof![10 => Just(Fir::Proper), 1 => Just(Fir::Flipped)]; + // a read's response mostly continues in another fragment, as a correct one must with CON + let fin_con = if read { + prop_oneof![ + 6 => Just((false, true)), + 3 => Just((true, false)), + 2 => Just((true, true)), + 1 => Just((false, false)), + ] + .boxed() + } else { + prop_oneof![ + 8 => Just((true, false)), + 4 => Just((true, true)), + 1 => Just((false, true)), + 1 => Just((false, false)), + ] + .boxed() + }; + let objects = if read { + measurement_objects().boxed() + } else { + response_objects().boxed() + }; + ( + source(), + seq, + fir, + fin_con, + prop::bool::weighted(0.15), + iin2(), + objects, + ) + .prop_map( + |(source, seq, fir, (fin, con), restart, iin2, objects)| Answer { + source, + seq, + fir, + fin, + con, + restart, + iin2, + objects, + }, + ) +} + +fn unsolicited() -> impl Strategy { + let source = prop_oneof![ + 3 => Just(Source::Outstation), + 1 => Just(Source::Other), + 1 => Just(Source::Unknown), + ]; + ( + source, + 0u8..16, + any::(), + prop::bool::weighted(0.15), + measurement_objects(), + prop::bool::weighted(0.3), + ) + .prop_map(|(source, seq, con, restart, objects, repeat)| Unsolicited { + source, + seq, + con, + restart, + objects, + repeat, + }) +} + +fn input(read: bool) -> impl Strategy { + // a malformed fragment from the outstation ends the task, so it comes from others more often + let malformed_source = prop_oneof![ + 1 => Just(Source::Outstation), + 2 => Just(Source::Other), + 2 => Just(Source::Unknown), + ]; + prop_oneof![ + 10 => answer(read).prop_map(Input::Answer), + 3 => unsolicited().prop_map(Input::Unsolicited), + 1 => (malformed_source, 0..MALFORMED.len()).prop_map(|(source, which)| Input::Malformed(source, which)), + 1 => Just(Input::Api), + 1 => Just(Input::Silence), + ] +} + +fn steps(read: bool) -> impl Strategy> { + prop::collection::vec( + (input(read), wait_ms()).prop_map(|(input, wait)| Step { input, wait }), + 1..=10, + ) +} + +proptest! { + // more cases than the other tasks, since few of them read several fragments + #![proptest_config(ProptestConfig::with_cases(128))] + + #[test] + fn read_tasks_match_the_model( + task in prop_oneof![Just(Task::User(Kind::Read)), Just(Task::StartupIntegrity)], + steps in steps(true), + ) { + run(task, &steps).check()?; + } +} + +proptest! { + #![proptest_config(ProptestConfig::with_cases(64))] + + #[test] + fn non_read_tasks_match_the_model( + kind in prop::sample::select(vec![ + Kind::DirectOperate, + Kind::SelectBeforeOperate, + Kind::ColdRestart, + Kind::EmptyResponse, + Kind::DeadBands, + Kind::AuthFile, + Kind::OpenFile, + Kind::CloseFile, + Kind::WriteFileBlock, + Kind::GetFileInfo, + ]), + steps in steps(false), + ) { + run(Task::User(kind), &steps).check()?; + } +} + +/// A read whose first fragment reports an unknown object but has data, and asks for a confirm, +/// then a last fragment without data +fn read_with_a_rejected_fragment() -> Vec { + let fragment = |fin, iin2, objects| Step { + input: Input::Answer(Answer { + source: Source::Outstation, + seq: SeqChoice::Expected, + fir: Fir::Proper, + fin, + con: true, + restart: false, + iin2, + objects, + }), + wait: TICK, + }; + vec![ + fragment(false, 0x02, Objects::AnalogEvent), + fragment(true, 0x00, Objects::None), + ] +} + +#[test] +fn the_model_accepts_the_master_and_rejects_known_wrong_outcomes() { + let task = Task::User(Kind::Read); + let mut run = run(task, &read_with_a_rejected_fragment()); + run.check().unwrap(); + + // the model's expectation, written out + let only = |allowed: &[Allowed]| { + assert_eq!(allowed.len(), 1); + allowed[0].obs.clone() + }; + let read = ReadType::SinglePoll; + assert_eq!( + only(&run.allowed[1]), + StepObs { + writes: vec![Write::Confirm(0)], + reads: 1, + main: vec![ + ReadEvent::BeginFragment(read), + ReadEvent::AnalogInput, + ReadEvent::EndFragment(read), + ], + ..StepObs::default() + } + ); + assert_eq!( + only(&run.allowed[2]), + StepObs { + writes: vec![Write::Confirm(1)], + reads: 1, + main: vec![ReadEvent::BeginFragment(read), ReadEvent::EndFragment(read)], + info: vec![AssocInfoEvent::TaskSuccess( + TaskType::UserRead, + FunctionCode::Read, + Sequence::new(1) + )], + ..StepObs::default() + } + ); + assert_eq!(run.reply, ReplyObs::Done(Reply::Read(Ok(())), 2 * TICK)); + assert_eq!( + run.probe, + StepObs { + writes: vec![Write::Request(FunctionCode::Read, 2)], + info: vec![AssocInfoEvent::TaskStart( + TaskType::UserRead, + FunctionCode::Read, + Sequence::new(2) + )], + ..StepObs::default() + } + ); + + // confirming the last fragment with the first fragment's sequence number + let observed = run.observed.clone(); + run.observed[2].writes = vec![Write::Confirm(0)]; + run.check().unwrap_err(); + run.observed = observed; + + // continuing the sequence numbers from the request rather than the last fragment + let probe = run.probe.clone(); + run.probe.writes = vec![Write::Request(FunctionCode::Read, 1)]; + run.check().unwrap_err(); + run.probe = probe; + + // failing the read for the request error, though it returned data + let rejected = TaskError::RejectedByIin2(Iin::new(Iin1::new(0), Iin2::new(0x02))); + run.observed[2].info = vec![AssocInfoEvent::TaskFailure(TaskType::UserRead, rejected)]; + run.reply = ReplyObs::Done(Reply::Read(Err(rejected)), 2 * TICK); + run.check().unwrap_err(); +} + +#[test] +fn a_startup_integrity_scan_times_out_one_response_timeout_after_its_last_fragment() { + let first = Step { + input: Input::Answer(Answer { + source: Source::Outstation, + seq: SeqChoice::Expected, + fir: Fir::Proper, + fin: false, + con: true, + restart: false, + iin2: 0, + objects: Objects::AnalogStatic, + }), + wait: FINAL_WAIT, + }; + let run = run(Task::StartupIntegrity, &[first]); + run.check().unwrap(); + + let failed = + AssocInfoEvent::TaskFailure(TaskType::StartupIntegrity, TaskError::ResponseTimeout); + // the fragment at 7 ms and nothing else until a millisecond before its timeout at 1007 ms, + // then the failure + assert_eq!(run.observed.len(), 3); + assert_eq!(run.observed[1].writes, vec![Write::Confirm(0)]); + assert_eq!(run.observed[1].info, vec![]); + assert_eq!( + run.observed[2], + StepObs { + info: vec![failed], + ..StepObs::default() + } + ); + assert_eq!(run.reply, ReplyObs::None); + + // timing out before 1007 ms, from the request at 0 ms + let mut early = run; + early.observed[1].info = vec![failed]; + early.observed[2].info = vec![]; + early.check().unwrap_err(); +} + +#[test] +fn the_model_rejects_a_missing_confirm_of_a_control_response() { + let task = Task::User(Kind::DirectOperate); + let response = Step { + input: Input::Answer(Answer { + source: Source::Outstation, + seq: SeqChoice::Expected, + fir: Fir::Proper, + fin: true, + con: true, + restart: false, + iin2: 0, + objects: Objects::Correct, + }), + wait: TICK, + }; + let mut run = run(task, &[response]); + run.check().unwrap(); + assert_eq!(run.observed[1].writes, vec![Write::Confirm(0)]); + assert_eq!(run.reply, ReplyObs::Done(Reply::Command(Ok(())), TICK)); + + run.observed[1].writes.clear(); + run.check().unwrap_err(); +}