private static void doTestSamzaPropertyAccess(String propertyName, String value, SamzaPropertyAssertion assertion) {
    // streams.<streamId>.<property>
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + propertyName, value,
        // all streams need to have a system
        JobConfig.JOB_DEFAULT_SYSTEM, SYSTEM))),
        SYSTEM_STREAM);

    // systems.<system>.streams.<stream>.<property> where stream id has no specified physical stream
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, value,
        // specify the system for the stream id
        String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))),
        SYSTEM_STREAM);

    // systems.<system>.streams.<stream>.<property> where stream is the streamId, system is from job default system
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, value,
        // use job default system to get the system
        JobConfig.JOB_DEFAULT_SYSTEM, SYSTEM))),
        SYSTEM_STREAM);

    // systems.<system>.default.stream.<property> where stream id has no specified physical stream
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(SystemConfig.SYSTEM_DEFAULT_STREAMS_PREFIX_FORMAT, SYSTEM) + propertyName, value,
        // specify the system for the stream id
        String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))),
        SYSTEM_STREAM);

    // systems.<system>.streams.<stream>.<property> where no system mapping (fall back to SystemStream)
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, value,
        // map the stream id to the physical stream
        String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM))),
        SYSTEM_STREAM);

    // systems.<system>.default.stream.<property> where no system mapping (fall back to SystemStream)
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(SystemConfig.SYSTEM_DEFAULT_STREAMS_PREFIX_FORMAT, SYSTEM) + propertyName, value,
        // map the stream id to the physical stream
        String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM))),
        SYSTEM_STREAM);

    // systems.<system>.streams.<stream>.<property> where stream id has a physical name but property is from stream id
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, value,
        // use job default system to get the system
        JobConfig.JOB_DEFAULT_SYSTEM, SYSTEM,
        // map the stream id to the physical stream
        String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM))),
        SYSTEM_STREAM);

    // systems.<system>.default.stream.<property> where stream id has a physical name but property is from stream id
    assertion.doAssertion(new StreamConfig(new MapConfig(ImmutableMap.of(
        String.format(SystemConfig.SYSTEM_DEFAULT_STREAMS_PREFIX_FORMAT, SYSTEM) + propertyName, value,
        // use job default system to get the system
        JobConfig.JOB_DEFAULT_SYSTEM, SYSTEM,
        // map the stream id to the physical stream
        String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM))),
        SYSTEM_STREAM);
  }
Tests for Samza property access using different lookup methods, when using stream id in system stream.
This includes tests in which there is no specified physical stream, so the stream id is used as the physical
stream.
samza-core/src/test/java/org/apache/samza/config/TestStreamConfig.java