text:
package org.apache.beam.runners.dataflow;

import static org.apache.beam.runners.dataflow.DataflowRunner.getContainerImageForJob;
import static org.hamcrest.Matchers.both;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyInt;
import static org.mockito.Matchers.anyListOf;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import com.google.api.services.dataflow.Dataflow;
import com.google.api.services.dataflow.model.DataflowPackage;
import com.google.api.services.dataflow.model.Job;
import com.google.api.services.dataflow.model.ListJobsResponse;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.Serializable;
import java.net.URL;
import java.net.URLClassLoader;
import java.nio.channels.FileChannel;
import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;
import org.apache.beam.runners.dataflow.DataflowRunner.StreamingShardedWriteFactory;
import org.apache.beam.runners.dataflow.options.DataflowPipelineDebugOptions;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.Pipeline.PipelineVisitor;
import org.apache.beam.sdk.coders.BigEndianIntegerCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.VoidCoder;
import org.apache.beam.sdk.extensions.gcp.auth.NoopCredentialFactory;
import org.apache.beam.sdk.extensions.gcp.auth.TestCredential;
import org.apache.beam.sdk.extensions.gcp.storage.NoopPathValidator;
import org.apache.beam.sdk.io.FileBasedSink;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.WriteFiles;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptions.CheckEnabled;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.runners.TransformHierarchy;
import org.apache.beam.sdk.runners.TransformHierarchy.Node;
import org.apache.beam.sdk.state.MapState;
import org.apache.beam.sdk.state.SetState;
import org.apache.beam.sdk.state.StateSpec;
import org.apache.beam.sdk.state.StateSpecs;
import org.apache.beam.sdk.state.ValueState;
import org.apache.beam.sdk.testing.ExpectedLogs;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.Sessions;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.util.GcsUtil;
import org.apache.beam.sdk.util.ReleaseInfo;
import org.apache.beam.sdk.util.gcsfs.GcsPath;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PDone;
import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TimestampedValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.hamcrest.Description;
import org.hamcrest.Matchers;
import org.hamcrest.TypeSafeMatcher;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

/**
 * Tests for the {@link DataflowRunner}.
 *
 * <p>Implements {@link Serializable} because it is caught in closures.
 */
@RunWith(JUnit4.class)
public class DataflowRunnerTest implements Serializable {

  private static final String VALID_STAGING_BUCKET = "gs://valid-bucket/staging";
  private static final String VALID_TEMP_BUCKET = "gs://valid-bucket/temp";
  private static final String VALID_PROFILE_BUCKET = "gs://valid-bucket/profiles";
  private static final String NON_EXISTENT_BUCKET = "gs://non-existent-bucket/location";

  private static final String PROJECT_ID = "some-project";
  private static final String REGION_ID = "some-region-1";

  @Rule public transient TemporaryFolder tmpFolder = new TemporaryFolder();
  @Rule public transient ExpectedException thrown = ExpectedException.none();
  @Rule public transient ExpectedLogs expectedLogs = ExpectedLogs.none(DataflowRunner.class);

  private transient Dataflow.Projects.Locations.Jobs mockJobs;
  private transient GcsUtil mockGcsUtil;

  // Asserts that the given Job has all expected fields set.
  private static void assertValidJob(Job job) {
    assertNull(job.getId());
    assertNull(job.getCurrentState());
    assertTrue(Pattern.matches("[a-z]([-a-z0-9]*[a-z0-9])?", job.getName()));
  }

  @Before
  public void setUp() throws IOException {
    this.mockGcsUtil = mock(GcsUtil.class);
    when(mockGcsUtil.create(any(GcsPath.class), anyString()))
        .then(new Answer<SeekableByteChannel>() {
          @Override
          public SeekableByteChannel answer(InvocationOnMock invocation) throws Throwable {
            return FileChannel.open(
                Files.createTempFile("channel-", ".tmp"),
                StandardOpenOption.CREATE, StandardOpenOption.DELETE_ON_CLOSE);
          }
        });
    when(mockGcsUtil.expand(any(GcsPath.class))).then(new Answer<List<GcsPath>>() {
      @Override
      public List<GcsPath> answer(InvocationOnMock invocation) throws Throwable {
        return ImmutableList.of((GcsPath) invocation.getArguments()[0]);
      }
    });
    when(mockGcsUtil.bucketAccessible(GcsPath.fromUri(VALID_STAGING_BUCKET))).thenReturn(true);
    when(mockGcsUtil.bucketAccessible(GcsPath.fromUri(VALID_STAGING_BUCKET))).thenReturn(true);
    when(mockGcsUtil.bucketAccessible(GcsPath.fromUri(VALID_TEMP_BUCKET))).thenReturn(true);
    when(mockGcsUtil.bucketAccessible(GcsPath.fromUri(VALID_TEMP_BUCKET + "/staging/"))).
        thenReturn(true);
    when(mockGcsUtil.bucketAccessible(GcsPath.fromUri(VALID_PROFILE_BUCKET))).thenReturn(true);
    when(mockGcsUtil.bucketAccessible(GcsPath.fromUri(NON_EXISTENT_BUCKET))).thenReturn(false);

    // The dataflow pipeline attempts to output to this location.
    when(mockGcsUtil.bucketAccessible(GcsPath.fromUri("gs://bucket/object"))).thenReturn(true);

    mockJobs = mock(Dataflow.Projects.Locations.Jobs.class);
  }

  private Pipeline buildDataflowPipeline(DataflowPipelineOptions options) {
    options.setStableUniqueNames(CheckEnabled.ERROR);
    options.setRunner(DataflowRunner.class);
    Pipeline p = Pipeline.create(options);

    p.apply("ReadMyFile", TextIO.read().from("gs://bucket/object"))
        .apply("WriteMyFile", TextIO.write().to("gs://bucket/object"));

    // Enable the FileSystems API to know about gs:// URIs in this test.
    FileSystems.setDefaultPipelineOptions(options);

    return p;
  }

  private Dataflow buildMockDataflow() throws IOException {
    Dataflow mockDataflowClient = mock(Dataflow.class);
    Dataflow.Projects mockProjects = mock(Dataflow.Projects.class);
    Dataflow.Projects.Locations mockLocations = mock(Dataflow.Projects.Locations.class);
    Dataflow.Projects.Locations.Jobs.Create mockRequest =
        mock(Dataflow.Projects.Locations.Jobs.Create.class);
    Dataflow.Projects.Locations.Jobs.List mockList = mock(
        Dataflow.Projects.Locations.Jobs.List.class);

    when(mockDataflowClient.projects()).thenReturn(mockProjects);
    when(mockProjects.locations()).thenReturn(mockLocations);
    when(mockLocations.jobs()).thenReturn(mockJobs);
    when(mockJobs.create(eq(PROJECT_ID), eq(REGION_ID), isA(Job.class))).thenReturn(mockRequest);
    when(mockJobs.list(eq(PROJECT_ID), eq(REGION_ID))).thenReturn(mockList);
    when(mockList.setPageToken(anyString())).thenReturn(mockList);
    when(mockList.execute())
        .thenReturn(
            new ListJobsResponse()
                .setJobs(
                    Arrays.asList(
                        new Job()
                            .setName("oldjobname")
                            .setId("oldJobId")
                            .setCurrentState("JOB_STATE_RUNNING"))));

    Job resultJob = new Job();
    resultJob.setId("newid");
    when(mockRequest.execute()).thenReturn(resultJob);
    return mockDataflowClient;
  }

  private GcsUtil buildMockGcsUtil() throws IOException {
    GcsUtil mockGcsUtil = mock(GcsUtil.class);
    when(mockGcsUtil.create(any(GcsPath.class), anyString()))
        .then(new Answer<SeekableByteChannel>() {
          @Override
          public SeekableByteChannel answer(InvocationOnMock invocation) throws Throwable {
            return FileChannel.open(
                Files.createTempFile("channel-", ".tmp"),
                StandardOpenOption.CREATE, StandardOpenOption.DELETE_ON_CLOSE);
          }
        });
    when(mockGcsUtil.expand(any(GcsPath.class))).then(new Answer<List<GcsPath>>() {
      @Override
      public List<GcsPath> answer(InvocationOnMock invocation) throws Throwable {
        return ImmutableList.of((GcsPath) invocation.getArguments()[0]);
      }
    });
    return mockGcsUtil;
  }

  private DataflowPipelineOptions buildPipelineOptions() throws IOException {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setProject(PROJECT_ID);
    options.setTempLocation(VALID_TEMP_BUCKET);
    options.setRegion(REGION_ID);
    // Set FILES_PROPERTY to empty to prevent a default value calculated from classpath.
    options.setFilesToStage(new LinkedList<String>());
    options.setDataflowClient(buildMockDataflow());
    options.setGcsUtil(mockGcsUtil);
    options.setGcpCredential(new TestCredential());

    // Configure the FileSystem registrar to use these options.
    FileSystems.setDefaultPipelineOptions(options);

    return options;
  }

  @Test
  public void testPathValidation() {
    String[] args = new String[] {
        "--runner=DataflowRunner",
        "--tempLocation=/tmp/not/a/gs/path",
        "--project=test-project",
        "--credentialFactoryClass=" + NoopCredentialFactory.class.getName(),
    };

    try {
      Pipeline.create(PipelineOptionsFactory.fromArgs(args).create()).run();
      fail();
    } catch (RuntimeException e) {
      assertThat(
          Throwables.getStackTraceAsString(e),
          containsString("DataflowRunner requires gcpTempLocation"));
    }
  }

  @Test
  public void testPathExistsValidation() {
    String[] args = new String[] {
        "--runner=DataflowRunner",
        "--tempLocation=gs://does/not/exist",
        "--project=test-project",
        "--credentialFactoryClass=" + NoopCredentialFactory.class.getName(),
    };

    try {
      Pipeline.create(PipelineOptionsFactory.fromArgs(args).create()).run();
      fail();
    } catch (RuntimeException e) {
      assertThat(
          Throwables.getStackTraceAsString(e),
          both(containsString("gs://does/not/exist"))
              .and(containsString("does not exist or is not writeable")));
    }
  }

  @Test
  public void testPathValidatorOverride() {
    String[] args = new String[] {
        "--runner=DataflowRunner",
        "--tempLocation=/tmp/testing",
        "--project=test-project",
        "--credentialFactoryClass=" + NoopCredentialFactory.class.getName(),
        "--pathValidatorClass=" + NoopPathValidator.class.getName(),
    };
    // Should not crash, because gcpTempLocation should get set from tempLocation
    TestPipeline.fromOptions(PipelineOptionsFactory.fromArgs(args).create());
  }

  @Test
  public void testFromOptionsWithUppercaseConvertsToLowercase() throws Exception {
    String mixedCase = "ThisJobNameHasMixedCase";
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setJobName(mixedCase);

    DataflowRunner.fromOptions(options);
    assertThat(options.getJobName(), equalTo(mixedCase.toLowerCase()));
  }

  @Test
  public void testRun() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    Pipeline p = buildDataflowPipeline(options);
    DataflowPipelineJob job = (DataflowPipelineJob) p.run();
    assertEquals("newid", job.getJobId());

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    assertValidJob(jobCaptor.getValue());
  }

  /** Options for testing. */
  public interface RuntimeTestOptions extends PipelineOptions {
    ValueProvider<String> getInput();
    void setInput(ValueProvider<String> value);

    ValueProvider<String> getOutput();
    void setOutput(ValueProvider<String> value);
  }

  @Test
  public void testTextIOWithRuntimeParameters() throws IOException {
    DataflowPipelineOptions dataflowOptions = buildPipelineOptions();
    RuntimeTestOptions options = dataflowOptions.as(RuntimeTestOptions.class);
    Pipeline p = buildDataflowPipeline(dataflowOptions);
    p
        .apply(TextIO.read().from(options.getInput()))
        .apply(TextIO.write().to(options.getOutput()));
  }

  /**
   * Tests that all reads are consumed by at least one {@link PTransform}.
   */
  @Test
  public void testUnconsumedReads() throws IOException {
    DataflowPipelineOptions dataflowOptions = buildPipelineOptions();
    RuntimeTestOptions options = dataflowOptions.as(RuntimeTestOptions.class);
    Pipeline p = buildDataflowPipeline(dataflowOptions);
    PCollection<String> unconsumed = p.apply(TextIO.read().from(options.getInput()));
    DataflowRunner.fromOptions(dataflowOptions).replaceTransforms(p);
    final AtomicBoolean unconsumedSeenAsInput = new AtomicBoolean();
    p.traverseTopologically(new PipelineVisitor.Defaults() {
      @Override
      public void visitPrimitiveTransform(Node node) {
        unconsumedSeenAsInput.set(true);
      }
    });
    assertThat(unconsumedSeenAsInput.get(), is(true));
  }

  @Test
  public void testRunReturnDifferentRequestId() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    Dataflow mockDataflowClient = options.getDataflowClient();
    Dataflow.Projects.Locations.Jobs.Create mockRequest = mock(
        Dataflow.Projects.Locations.Jobs.Create.class);
    when(mockDataflowClient.projects().locations().jobs()
        .create(eq(PROJECT_ID), eq(REGION_ID), any(Job.class)))
        .thenReturn(mockRequest);
    Job resultJob = new Job();
    resultJob.setId("newid");
    // Return a different request id.
    resultJob.setClientRequestId("different_request_id");
    when(mockRequest.execute()).thenReturn(resultJob);

    Pipeline p = buildDataflowPipeline(options);
    try {
      p.run();
      fail("Expected DataflowJobAlreadyExistsException");
    } catch (DataflowJobAlreadyExistsException expected) {
      assertThat(expected.getMessage(),
          containsString("If you want to submit a second job, try again by setting a "
              + "different name using --jobName."));
      assertEquals(expected.getJob().getJobId(), resultJob.getId());
    }
  }

  @Test
  public void testUpdate() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setUpdate(true);
    options.setJobName("oldJobName");
    Pipeline p = buildDataflowPipeline(options);
    DataflowPipelineJob job = (DataflowPipelineJob) p.run();
    assertEquals("newid", job.getJobId());

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    assertValidJob(jobCaptor.getValue());
  }

  @Test
  public void testUpdateNonExistentPipeline() throws IOException {
    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("Could not find running job named badjobname");

    DataflowPipelineOptions options = buildPipelineOptions();
    options.setUpdate(true);
    options.setJobName("badJobName");
    Pipeline p = buildDataflowPipeline(options);
    p.run();
  }

  @Test
  public void testUpdateAlreadyUpdatedPipeline() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setUpdate(true);
    options.setJobName("oldJobName");
    Dataflow mockDataflowClient = options.getDataflowClient();
    Dataflow.Projects.Locations.Jobs.Create mockRequest = mock(
        Dataflow.Projects.Locations.Jobs.Create.class);
    when(mockDataflowClient.projects().locations().jobs()
        .create(eq(PROJECT_ID), eq(REGION_ID), any(Job.class)))
        .thenReturn(mockRequest);
    final Job resultJob = new Job();
    resultJob.setId("newid");
    // Return a different request id.
    resultJob.setClientRequestId("different_request_id");
    when(mockRequest.execute()).thenReturn(resultJob);

    Pipeline p = buildDataflowPipeline(options);

    thrown.expect(DataflowJobAlreadyUpdatedException.class);
    thrown.expect(new TypeSafeMatcher<DataflowJobAlreadyUpdatedException>() {
      @Override
      public void describeTo(Description description) {
        description.appendText("Expected job ID: " + resultJob.getId());
      }

      @Override
      protected boolean matchesSafely(DataflowJobAlreadyUpdatedException item) {
        return resultJob.getId().equals(item.getJob().getJobId());
      }
    });
    thrown.expectMessage("The job named oldjobname with id: oldJobId has already been updated "
        + "into job id: newid and cannot be updated again.");
    p.run();
  }

  @Test
  public void testRunWithFiles() throws IOException {
    // Test that the function DataflowRunner.stageFiles works as expected.
    final String cloudDataflowDataset = "somedataset";

    // Create some temporary files.
    File temp1 = File.createTempFile("DataflowRunnerTest", "txt");
    temp1.deleteOnExit();
    File temp2 = File.createTempFile("DataflowRunnerTest2", "txt");
    temp2.deleteOnExit();

    String overridePackageName = "alias.txt";

    when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
        .thenReturn(ImmutableList.of(GcsUtil.StorageObjectOrIOException.create(
            new FileNotFoundException("some/path"))));

    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setFilesToStage(ImmutableList.of(
        temp1.getAbsolutePath(),
        overridePackageName + "=" + temp2.getAbsolutePath()));
    options.setStagingLocation(VALID_STAGING_BUCKET);
    options.setTempLocation(VALID_TEMP_BUCKET);
    options.setTempDatasetId(cloudDataflowDataset);
    options.setProject(PROJECT_ID);
    options.setRegion(REGION_ID);
    options.setJobName("job");
    options.setDataflowClient(buildMockDataflow());
    options.setGcsUtil(mockGcsUtil);
    options.setGcpCredential(new TestCredential());

    when(mockGcsUtil.create(any(GcsPath.class), anyString(), anyInt()))
        .then(new Answer<SeekableByteChannel>() {
          @Override
          public SeekableByteChannel answer(InvocationOnMock invocation) throws Throwable {
            return FileChannel.open(
                Files.createTempFile("channel-", ".tmp"),
                StandardOpenOption.CREATE, StandardOpenOption.DELETE_ON_CLOSE);
          }
        });

    Pipeline p = buildDataflowPipeline(options);

    DataflowPipelineJob job = (DataflowPipelineJob) p.run();
    assertEquals("newid", job.getJobId());

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    Job workflowJob = jobCaptor.getValue();
    assertValidJob(workflowJob);

    assertEquals(
        2,
        workflowJob.getEnvironment().getWorkerPools().get(0).getPackages().size());
    DataflowPackage workflowPackage1 =
        workflowJob.getEnvironment().getWorkerPools().get(0).getPackages().get(0);
    assertThat(workflowPackage1.getName(), startsWith(temp1.getName()));
    DataflowPackage workflowPackage2 =
        workflowJob.getEnvironment().getWorkerPools().get(0).getPackages().get(1);
    assertEquals(overridePackageName, workflowPackage2.getName());

    assertEquals(
        GcsPath.fromUri(VALID_TEMP_BUCKET).toResourceName(),
        workflowJob.getEnvironment().getTempStoragePrefix());
    assertEquals(
        cloudDataflowDataset,
        workflowJob.getEnvironment().getDataset());
    assertEquals(
        ReleaseInfo.getReleaseInfo().getName(),
        workflowJob.getEnvironment().getUserAgent().get("name"));
    assertEquals(
        ReleaseInfo.getReleaseInfo().getVersion(),
        workflowJob.getEnvironment().getUserAgent().get("version"));
  }

  @Test
  public void runWithDefaultFilesToStage() throws Exception {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setFilesToStage(null);
    DataflowRunner.fromOptions(options);
    assertTrue(!options.getFilesToStage().isEmpty());
  }

  @Test
  public void detectClassPathResourceWithFileResources() throws Exception {
    File file = tmpFolder.newFile("file");
    File file2 = tmpFolder.newFile("file2");
    URLClassLoader classLoader = new URLClassLoader(new URL[] {
        file.toURI().toURL(),
        file2.toURI().toURL()
    });

    assertEquals(ImmutableList.of(file.getAbsolutePath(), file2.getAbsolutePath()),
        DataflowRunner.detectClassPathResourcesToStage(classLoader));
  }

  @Test
  public void detectClassPathResourcesWithUnsupportedClassLoader() {
    ClassLoader mockClassLoader = Mockito.mock(ClassLoader.class);
    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("Unable to use ClassLoader to detect classpath elements.");

    DataflowRunner.detectClassPathResourcesToStage(mockClassLoader);
  }

  @Test
  public void detectClassPathResourceWithNonFileResources() throws Exception {
    String url = "http://www.google.com/all-the-secrets.jar";
    URLClassLoader classLoader = new URLClassLoader(new URL[] {
        new URL(url)
    });
    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("Unable to convert url (" + url + ") to file.");

    DataflowRunner.detectClassPathResourcesToStage(classLoader);
  }

  @Test
  public void testGcsStagingLocationInitialization() throws Exception {
    // Set temp location (required), and check that staging location is set.
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setTempLocation(VALID_TEMP_BUCKET);
    options.setProject(PROJECT_ID);
    options.setGcpCredential(new TestCredential());
    options.setGcsUtil(mockGcsUtil);
    options.setRunner(DataflowRunner.class);

    DataflowRunner.fromOptions(options);

    assertNotNull(options.getStagingLocation());
  }

  @Test
  public void testInvalidGcpTempLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setGcpTempLocation("file://temp/location");

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(containsString("Expected a valid 'gs://' path but was given"));
    DataflowRunner.fromOptions(options);

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    assertValidJob(jobCaptor.getValue());
  }

  @Test
  public void testNonGcsTempLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setTempLocation("file://temp/location");

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(
        "DataflowRunner requires gcpTempLocation, "
            + "but failed to retrieve a value from PipelineOptions");
    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testInvalidStagingLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setStagingLocation("file://my/staging/location");
    try {
      DataflowRunner.fromOptions(options);
      fail("fromOptions should have failed");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage(), containsString("Expected a valid 'gs://' path but was given"));
    }
    options.setStagingLocation("my/staging/location");
    try {
      DataflowRunner.fromOptions(options);
      fail("fromOptions should have failed");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage(), containsString("Expected a valid 'gs://' path but was given"));
    }
  }

  @Test
  public void testInvalidProfileLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setSaveProfilesToGcs("file://my/staging/location");
    try {
      DataflowRunner.fromOptions(options);
      fail("fromOptions should have failed");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage(), containsString("Expected a valid 'gs://' path but was given"));
    }
    options.setSaveProfilesToGcs("my/staging/location");
    try {
      DataflowRunner.fromOptions(options);
      fail("fromOptions should have failed");
    } catch (IllegalArgumentException e) {
      assertThat(e.getMessage(), containsString("Expected a valid 'gs://' path but was given"));
    }
  }

  @Test
  public void testNonExistentTempLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setGcpTempLocation(NON_EXISTENT_BUCKET);

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(containsString(
        "Output path does not exist or is not writeable: " + NON_EXISTENT_BUCKET));
    DataflowRunner.fromOptions(options);

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    assertValidJob(jobCaptor.getValue());
  }

  @Test
  public void testNonExistentStagingLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setStagingLocation(NON_EXISTENT_BUCKET);

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(containsString(
        "Output path does not exist or is not writeable: " + NON_EXISTENT_BUCKET));
    DataflowRunner.fromOptions(options);

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    assertValidJob(jobCaptor.getValue());
  }

  @Test
  public void testNonExistentProfileLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setSaveProfilesToGcs(NON_EXISTENT_BUCKET);

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(containsString(
        "Output path does not exist or is not writeable: " + NON_EXISTENT_BUCKET));
    DataflowRunner.fromOptions(options);

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    assertValidJob(jobCaptor.getValue());
  }

  @Test
  public void testNoProjectFails() {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);

    options.setRunner(DataflowRunner.class);
    // Explicitly set to null to prevent the default instance factory from reading credentials
    // from a user's environment, causing this test to fail.
    options.setProject(null);

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("Project id");
    thrown.expectMessage("when running a Dataflow in the cloud");

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testProjectId() throws IOException {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setProject("foo-12345");

    options.setGcpTempLocation(VALID_TEMP_BUCKET);
    options.setGcsUtil(mockGcsUtil);
    options.setGcpCredential(new TestCredential());

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testProjectPrefix() throws IOException {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setProject("google.com:some-project-12345");

    options.setGcpTempLocation(VALID_TEMP_BUCKET);
    options.setGcsUtil(mockGcsUtil);
    options.setGcpCredential(new TestCredential());

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testProjectNumber() throws IOException {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setProject("12345");

    options.setGcpTempLocation(VALID_TEMP_BUCKET);
    options.setGcsUtil(mockGcsUtil);

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("Project ID");
    thrown.expectMessage("project number");

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testProjectDescription() throws IOException {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setProject("some project");

    options.setGcpTempLocation(VALID_TEMP_BUCKET);
    options.setGcsUtil(mockGcsUtil);

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("Project ID");
    thrown.expectMessage("project description");

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testInvalidNumberOfWorkerHarnessThreads() throws IOException {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    FileSystems.setDefaultPipelineOptions(options);
    options.setRunner(DataflowRunner.class);
    options.setProject("foo-12345");

    options.setGcpTempLocation(VALID_TEMP_BUCKET);
    options.setGcsUtil(mockGcsUtil);

    options.as(DataflowPipelineDebugOptions.class).setNumberOfWorkerHarnessThreads(-1);

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("Number of worker harness threads");
    thrown.expectMessage("Please make sure the value is non-negative.");

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testNoStagingLocationAndNoTempLocationFails() {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setProject("foo-project");

    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("DataflowRunner requires gcpTempLocation, "
        + "but failed to retrieve a value from PipelineOption");
    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testGcpTempAndNoTempLocationSucceeds() throws Exception {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setGcpCredential(new TestCredential());
    options.setProject("foo-project");
    options.setGcpTempLocation(VALID_TEMP_BUCKET);
    options.setGcsUtil(mockGcsUtil);

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testTempLocationAndNoGcpTempLocationSucceeds() throws Exception {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setRunner(DataflowRunner.class);
    options.setGcpCredential(new TestCredential());
    options.setProject("foo-project");
    options.setTempLocation(VALID_TEMP_BUCKET);
    options.setGcsUtil(mockGcsUtil);

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testValidProfileLocation() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    options.setSaveProfilesToGcs(VALID_PROFILE_BUCKET);

    DataflowRunner.fromOptions(options);
  }

  @Test
  public void testInvalidJobName() throws IOException {
    List<String> invalidNames = Arrays.asList(
        "invalid_name",
        "0invalid",
        "invalid-");
    List<String> expectedReason = Arrays.asList(
        "JobName invalid",
        "JobName invalid",
        "JobName invalid");

    for (int i = 0; i < invalidNames.size(); ++i) {
      DataflowPipelineOptions options = buildPipelineOptions();
      options.setJobName(invalidNames.get(i));

      try {
        DataflowRunner.fromOptions(options);
        fail("Expected IllegalArgumentException for jobName "
            + options.getJobName());
      } catch (IllegalArgumentException e) {
        assertThat(e.getMessage(),
            containsString(expectedReason.get(i)));
      }
    }
  }

  @Test
  public void testValidJobName() throws IOException {
    List<String> names = Arrays.asList("ok", "Ok", "A-Ok", "ok-123",
        "this-one-is-fairly-long-01234567890123456789");

    for (String name : names) {
      DataflowPipelineOptions options = buildPipelineOptions();
      options.setJobName(name);

      DataflowRunner runner = DataflowRunner
          .fromOptions(options);
      assertNotNull(runner);
    }
  }

  @Test
  public void testGcsUploadBufferSizeIsUnsetForBatchWhenDefault() throws IOException {
    DataflowPipelineOptions batchOptions = buildPipelineOptions();
    batchOptions.setRunner(DataflowRunner.class);
    Pipeline.create(batchOptions);
    assertNull(batchOptions.getGcsUploadBufferSizeBytes());
  }

  @Test
  public void testGcsUploadBufferSizeIsSetForStreamingWhenDefault() throws IOException {
    DataflowPipelineOptions streamingOptions = buildPipelineOptions();
    streamingOptions.setStreaming(true);
    streamingOptions.setRunner(DataflowRunner.class);
    Pipeline p = Pipeline.create(streamingOptions);

    // Instantiation of a runner prior to run() currently has a side effect of mutating the options.
    // This could be tested by DataflowRunner.fromOptions(streamingOptions) but would not ensure
    // that the pipeline itself had the expected options set.
    p.run();

    assertEquals(
        DataflowRunner.GCS_UPLOAD_BUFFER_SIZE_BYTES_DEFAULT,
        streamingOptions.getGcsUploadBufferSizeBytes().intValue());
  }

  @Test
  public void testGcsUploadBufferSizeUnchangedWhenNotDefault() throws IOException {
    int gcsUploadBufferSizeBytes = 12345678;
    DataflowPipelineOptions batchOptions = buildPipelineOptions();
    batchOptions.setGcsUploadBufferSizeBytes(gcsUploadBufferSizeBytes);
    batchOptions.setRunner(DataflowRunner.class);
    Pipeline.create(batchOptions);
    assertEquals(gcsUploadBufferSizeBytes, batchOptions.getGcsUploadBufferSizeBytes().intValue());

    DataflowPipelineOptions streamingOptions = buildPipelineOptions();
    streamingOptions.setStreaming(true);
    streamingOptions.setGcsUploadBufferSizeBytes(gcsUploadBufferSizeBytes);
    streamingOptions.setRunner(DataflowRunner.class);
    Pipeline.create(streamingOptions);
    assertEquals(
        gcsUploadBufferSizeBytes, streamingOptions.getGcsUploadBufferSizeBytes().intValue());
  }

  /**
   * A fake PTransform for testing.
   */
  public static class TestTransform
      extends PTransform<PCollection<Integer>, PCollection<Integer>> {
    public boolean translated = false;

    @Override
    public PCollection<Integer> expand(PCollection<Integer> input) {
      return PCollection.<Integer>createPrimitiveOutputInternal(
          input.getPipeline(),
          WindowingStrategy.globalDefault(),
          input.isBounded());
    }

    @Override
    protected Coder<?> getDefaultOutputCoder(PCollection<Integer> input) {
      return input.getCoder();
    }
  }

  @Test
  public void testTransformTranslatorMissing() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    Pipeline p = Pipeline.create(options);

    p.apply(Create.of(Arrays.asList(1, 2, 3)))
        .apply(new TestTransform());

    thrown.expect(IllegalStateException.class);
    thrown.expectMessage(Matchers.containsString("no translator registered"));
    DataflowPipelineTranslator.fromOptions(options)
        .translate(
            p, DataflowRunner.fromOptions(options), Collections.<DataflowPackage>emptyList());

    ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
    Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture());
    assertValidJob(jobCaptor.getValue());
  }

  @Test
  public void testTransformTranslator() throws IOException {
    // Test that we can provide a custom translation
    DataflowPipelineOptions options = buildPipelineOptions();
    Pipeline p = Pipeline.create(options);
    TestTransform transform = new TestTransform();

    p.apply(Create.of(Arrays.asList(1, 2, 3)).withCoder(BigEndianIntegerCoder.of()))
        .apply(transform);

    DataflowPipelineTranslator translator = DataflowRunner
        .fromOptions(options).getTranslator();

    DataflowPipelineTranslator.registerTransformTranslator(
        TestTransform.class,
        new TransformTranslator<TestTransform>() {
          @SuppressWarnings("unchecked")
          @Override
          public void translate(
              TestTransform transform,
              TranslationContext context) {
            transform.translated = true;

            // Note: This is about the minimum needed to fake out a
            // translation. This obviously isn't a real translation.
            StepTranslationContext stepContext = context.addStep(transform, "TestTranslate");
            stepContext.addOutput(context.getOutput(transform));
          }
        });

    translator.translate(
        p, DataflowRunner.fromOptions(options), Collections.<DataflowPackage>emptyList());
    assertTrue(transform.translated);
  }

  private void verifyMapStateUnsupported(PipelineOptions options) throws Exception {
    Pipeline p = Pipeline.create(options);
    p.apply(Create.of(KV.of(13, 42)))
        .apply(
            ParDo.of(
                new DoFn<KV<Integer, Integer>, Void>() {
                  @StateId("fizzle")
                  private final StateSpec<MapState<Void, Void>> voidState = StateSpecs.map();

                  @ProcessElement
                  public void process() {}
                }));

    thrown.expectMessage("MapState");
    thrown.expect(UnsupportedOperationException.class);
    p.run();
  }

  @Test
  public void testMapStateUnsupportedInBatch() throws Exception {
    PipelineOptions options = buildPipelineOptions();
    options.as(StreamingOptions.class).setStreaming(false);
    verifyMapStateUnsupported(options);
  }

  @Test
  public void testMapStateUnsupportedInStreaming() throws Exception {
    PipelineOptions options = buildPipelineOptions();
    options.as(StreamingOptions.class).setStreaming(true);
    verifyMapStateUnsupported(options);
  }

  private void verifySetStateUnsupported(PipelineOptions options) throws Exception {
    Pipeline p = Pipeline.create(options);
    p.apply(Create.of(KV.of(13, 42)))
        .apply(
            ParDo.of(
                new DoFn<KV<Integer, Integer>, Void>() {
                  @StateId("fizzle")
                  private final StateSpec<SetState<Void>> voidState = StateSpecs.set();

                  @ProcessElement
                  public void process() {}
                }));

    thrown.expectMessage("SetState");
    thrown.expect(UnsupportedOperationException.class);
    p.run();
  }

  @Test
  public void testSetStateUnsupportedInBatch() throws Exception {
    PipelineOptions options = buildPipelineOptions();
    options.as(StreamingOptions.class).setStreaming(false);
    Pipeline p = Pipeline.create(options);
    verifySetStateUnsupported(options);
  }

  @Test
  public void testSetStateUnsupportedInStreaming() throws Exception {
    PipelineOptions options = buildPipelineOptions();
    options.as(StreamingOptions.class).setStreaming(true);
    verifySetStateUnsupported(options);
  }

  /** Records all the composite transforms visited within the Pipeline. */
  private static class CompositeTransformRecorder extends PipelineVisitor.Defaults {
    private List<PTransform<?, ?>> transforms = new ArrayList<>();

    @Override
    public CompositeBehavior enterCompositeTransform(TransformHierarchy.Node node) {
      if (node.getTransform() != null) {
        transforms.add(node.getTransform());
      }
      return CompositeBehavior.ENTER_TRANSFORM;
    }

    public List<PTransform<?, ?>> getCompositeTransforms() {
      return transforms;
    }
  }

  @Test
  public void testApplyIsScopedToExactClass() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    Pipeline p = Pipeline.create(options);

    Create.TimestampedValues<String> transform =
        Create.timestamped(Arrays.asList(TimestampedValue.of("TestString", Instant.now())));
    p.apply(transform);

    CompositeTransformRecorder recorder = new CompositeTransformRecorder();
    p.traverseTopologically(recorder);

    // The recorder will also have seen a Create.Values composite as well, but we can't obtain that
    // transform.
    assertThat(
        "Expected to have seen CreateTimestamped composite transform.",
        recorder.getCompositeTransforms(),
        hasItem(transform));
    assertThat(
        "Expected to have two composites, CreateTimestamped and Create.Values",
        recorder.getCompositeTransforms(),
        hasItem(Matchers.<PTransform<?, ?>>isA((Class) Create.Values.class)));
  }

  @Test
  public void testToString() {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setJobName("TestJobName");
    options.setProject("test-project");
    options.setTempLocation("gs://test/temp/location");
    options.setGcpCredential(new TestCredential());
    options.setPathValidatorClass(NoopPathValidator.class);
    options.setRunner(DataflowRunner.class);
    assertEquals(
        "DataflowRunner#testjobname",
        DataflowRunner.fromOptions(options).toString());
  }

  /**
   * Tests that the {@link DataflowRunner} with {@code --templateLocation} returns normally when the
   * runner is successfully run.
   */
  @Test
  public void testTemplateRunnerFullCompletion() throws Exception {
    File existingFile = tmpFolder.newFile();
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setJobName("TestJobName");
    options.setGcpCredential(new TestCredential());
    options.setPathValidatorClass(NoopPathValidator.class);
    options.setProject("test-project");
    options.setRunner(DataflowRunner.class);
    options.setTemplateLocation(existingFile.getPath());
    options.setTempLocation(tmpFolder.getRoot().getPath());
    Pipeline p = Pipeline.create(options);

    p.run();
    expectedLogs.verifyInfo("Template successfully created");
  }

  /**
   * Tests that the {@link DataflowRunner} with {@code --templateLocation} throws the appropriate
   * exception when an output file is not writable.
   */
  @Test
  public void testTemplateRunnerLoggedErrorForFile() throws Exception {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setJobName("TestJobName");
    options.setRunner(DataflowRunner.class);
    options.setTemplateLocation("//bad/path");
    options.setProject("test-project");
    options.setTempLocation(tmpFolder.getRoot().getPath());
    options.setGcpCredential(new TestCredential());
    options.setPathValidatorClass(NoopPathValidator.class);
    Pipeline p = Pipeline.create(options);

    thrown.expectMessage("Cannot create output file at");
    thrown.expect(RuntimeException.class);
    p.run();
  }

  @Test
  public void testHasExperiment() {
    DataflowPipelineDebugOptions options =
        PipelineOptionsFactory.as(DataflowPipelineDebugOptions.class);

    options.setExperiments(null);
    assertFalse(DataflowRunner.hasExperiment(options, "foo"));

    options.setExperiments(ImmutableList.of("foo", "bar"));
    assertTrue(DataflowRunner.hasExperiment(options, "foo"));
    assertTrue(DataflowRunner.hasExperiment(options, "bar"));
    assertFalse(DataflowRunner.hasExperiment(options, "baz"));
    assertFalse(DataflowRunner.hasExperiment(options, "ba"));
    assertFalse(DataflowRunner.hasExperiment(options, "BAR"));
  }

  @Test
  public void testWorkerHarnessContainerImage() {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);

    // default image set
    options.setWorkerHarnessContainerImage("some-container");
    assertThat(getContainerImageForJob(options), equalTo("some-container"));

    // batch, legacy
    options.setWorkerHarnessContainerImage("gcr.io/IMAGE/foo");
    options.setExperiments(null);
    options.setStreaming(false);
    assertThat(
        getContainerImageForJob(options), equalTo("gcr.io/beam-java-batch/foo"));
    // streaming, legacy
    options.setStreaming(true);
    assertThat(
        getContainerImageForJob(options), equalTo("gcr.io/beam-java-streaming/foo"));
    // streaming, fnapi
    options.setExperiments(ImmutableList.of("experiment1", "beam_fn_api"));
    assertThat(
        getContainerImageForJob(options), equalTo("gcr.io/java/foo"));
  }

  @Test
  public void testStreamingWriteWithNoShardingReturnsNewTransform() {
    PipelineOptions options = TestPipeline.testingPipelineOptions();
    options.as(DataflowPipelineWorkerPoolOptions.class).setMaxNumWorkers(10);
    testStreamingWriteOverride(options, 20);
  }

  @Test
  public void testStreamingWriteWithNoShardingReturnsNewTransformMaxWorkersUnset() {
    PipelineOptions options = TestPipeline.testingPipelineOptions();
    testStreamingWriteOverride(options, StreamingShardedWriteFactory.DEFAULT_NUM_SHARDS);
  }

  private void verifyMergingStatefulParDoRejected(PipelineOptions options) throws Exception {
    Pipeline p = Pipeline.create(options);

    p.apply(Create.of(KV.of(13, 42)))
        .apply(Window.<KV<Integer, Integer>>into(Sessions.withGapDuration(Duration.millis(1))))
        .apply(ParDo.of(new DoFn<KV<Integer, Integer>, Void>() {
          @StateId("fizzle")
          private final StateSpec<ValueState<Void>> voidState = StateSpecs.value();

          @ProcessElement
          public void process() {}
        }));

    thrown.expectMessage("merging");
    thrown.expect(UnsupportedOperationException.class);
    p.run();
  }

  @Test
  public void testMergingStatefulRejectedInStreaming() throws Exception {
    PipelineOptions options = buildPipelineOptions();
    options.as(StreamingOptions.class).setStreaming(true);
    verifyMergingStatefulParDoRejected(options);
  }

  @Test
  public void testMergingStatefulRejectedInBatch() throws Exception {
    PipelineOptions options = buildPipelineOptions();
    options.as(StreamingOptions.class).setStreaming(false);
    verifyMergingStatefulParDoRejected(options);
  }

  private void testStreamingWriteOverride(PipelineOptions options, int expectedNumShards) {
    TestPipeline p = TestPipeline.fromOptions(options);

    StreamingShardedWriteFactory<Object> factory =
        new StreamingShardedWriteFactory<>(p.getOptions());
    WriteFiles<Object> original = WriteFiles.to(new TestSink(tmpFolder.toString()));
    PCollection<Object> objs = (PCollection) p.apply(Create.empty(VoidCoder.of()));
    AppliedPTransform<PCollection<Object>, PDone, WriteFiles<Object>> originalApplication =
        AppliedPTransform.of(
            "writefiles", objs.expand(), Collections.<TupleTag<?>, PValue>emptyMap(), original, p);

    WriteFiles<Object> replacement = (WriteFiles<Object>)
        factory.getReplacementTransform(originalApplication).getTransform();
    assertThat(replacement, not(equalTo((Object) original)));
    assertThat(replacement.getNumShards().get(), equalTo(expectedNumShards));
  }

  private static class TestSink extends FileBasedSink<Object> {
    @Override
    public void validate(PipelineOptions options) {}

    TestSink(String tmpFolder) {
      super(StaticValueProvider.of(FileSystems.matchNewResource(tmpFolder, true)),
          null);
    }
    @Override
    public WriteOperation<Object> createWriteOperation() {
      throw new IllegalArgumentException("Should not be used");
    }
  }
}


text:
#include "dcmtk/config/osconfig.h"     // make sure OS specific configuration is included first

#include "dcmtk/dcmrt/seq/drtss.h"

// --- item class ---

DRTSourceSequence::Item::Item(const OFBool emptyDefaultItem)
  : EmptyDefaultItem(emptyDefaultItem),
    ActiveSourceDiameter(DCM_ActiveSourceDiameter),
    ActiveSourceLength(DCM_ActiveSourceLength),
    MaterialID(DCM_MaterialID),
    ReferenceAirKermaRate(DCM_ReferenceAirKermaRate),
    SourceDescription(DCM_SourceDescription),
    SourceEncapsulationNominalThickness(DCM_SourceEncapsulationNominalThickness),
    SourceEncapsulationNominalTransmission(DCM_SourceEncapsulationNominalTransmission),
    SourceIsotopeHalfLife(DCM_SourceIsotopeHalfLife),
    SourceIsotopeName(DCM_SourceIsotopeName),
    SourceManufacturer(DCM_SourceManufacturer),
    SourceModelID(DCM_SourceModelID),
    SourceNumber(DCM_SourceNumber),
    SourceSerialNumber(DCM_SourceSerialNumber),
    SourceStrength(DCM_SourceStrength),
    SourceStrengthReferenceDate(DCM_SourceStrengthReferenceDate),
    SourceStrengthReferenceTime(DCM_SourceStrengthReferenceTime),
    SourceStrengthUnits(DCM_SourceStrengthUnits),
    SourceType(DCM_SourceType)
{
}

DRTSourceSequence::Item::Item(const Item &copy)
  : EmptyDefaultItem(copy.EmptyDefaultItem),
    ActiveSourceDiameter(copy.ActiveSourceDiameter),
    ActiveSourceLength(copy.ActiveSourceLength),
    MaterialID(copy.MaterialID),
    ReferenceAirKermaRate(copy.ReferenceAirKermaRate),
    SourceDescription(copy.SourceDescription),
    SourceEncapsulationNominalThickness(copy.SourceEncapsulationNominalThickness),
    SourceEncapsulationNominalTransmission(copy.SourceEncapsulationNominalTransmission),
    SourceIsotopeHalfLife(copy.SourceIsotopeHalfLife),
    SourceIsotopeName(copy.SourceIsotopeName),
    SourceManufacturer(copy.SourceManufacturer),
    SourceModelID(copy.SourceModelID),
    SourceNumber(copy.SourceNumber),
    SourceSerialNumber(copy.SourceSerialNumber),
    SourceStrength(copy.SourceStrength),
    SourceStrengthReferenceDate(copy.SourceStrengthReferenceDate),
    SourceStrengthReferenceTime(copy.SourceStrengthReferenceTime),
    SourceStrengthUnits(copy.SourceStrengthUnits),
    SourceType(copy.SourceType)
{
}

DRTSourceSequence::Item::~Item()
{
}

DRTSourceSequence::Item &DRTSourceSequence::Item::operator=(const Item &copy)
{
    if (this != &copy)
    {
        EmptyDefaultItem = copy.EmptyDefaultItem;
        ActiveSourceDiameter = copy.ActiveSourceDiameter;
        ActiveSourceLength = copy.ActiveSourceLength;
        MaterialID = copy.MaterialID;
        ReferenceAirKermaRate = copy.ReferenceAirKermaRate;
        SourceDescription = copy.SourceDescription;
        SourceEncapsulationNominalThickness = copy.SourceEncapsulationNominalThickness;
        SourceEncapsulationNominalTransmission = copy.SourceEncapsulationNominalTransmission;
        SourceIsotopeHalfLife = copy.SourceIsotopeHalfLife;
        SourceIsotopeName = copy.SourceIsotopeName;
        SourceManufacturer = copy.SourceManufacturer;
        SourceModelID = copy.SourceModelID;
        SourceNumber = copy.SourceNumber;
        SourceSerialNumber = copy.SourceSerialNumber;
        SourceStrength = copy.SourceStrength;
        SourceStrengthReferenceDate = copy.SourceStrengthReferenceDate;
        SourceStrengthReferenceTime = copy.SourceStrengthReferenceTime;
        SourceStrengthUnits = copy.SourceStrengthUnits;
        SourceType = copy.SourceType;
    }
    return *this;
}

void DRTSourceSequence::Item::clear()
{
    if (!EmptyDefaultItem)
    {
        /* clear all DICOM attributes */
        SourceNumber.clear();
        SourceSerialNumber.clear();
        SourceModelID.clear();
        SourceDescription.clear();
        SourceType.clear();
        SourceManufacturer.clear();
        ActiveSourceDiameter.clear();
        ActiveSourceLength.clear();
        MaterialID.clear();
        SourceEncapsulationNominalThickness.clear();
        SourceEncapsulationNominalTransmission.clear();
        SourceIsotopeName.clear();
        SourceIsotopeHalfLife.clear();
        SourceStrengthUnits.clear();
        ReferenceAirKermaRate.clear();
        SourceStrength.clear();
        SourceStrengthReferenceDate.clear();
        SourceStrengthReferenceTime.clear();
    }
}

OFBool DRTSourceSequence::Item::isEmpty()
{
    return SourceNumber.isEmpty() &&
           SourceSerialNumber.isEmpty() &&
           SourceModelID.isEmpty() &&
           SourceDescription.isEmpty() &&
           SourceType.isEmpty() &&
           SourceManufacturer.isEmpty() &&
           ActiveSourceDiameter.isEmpty() &&
           ActiveSourceLength.isEmpty() &&
           MaterialID.isEmpty() &&
           SourceEncapsulationNominalThickness.isEmpty() &&
           SourceEncapsulationNominalTransmission.isEmpty() &&
           SourceIsotopeName.isEmpty() &&
           SourceIsotopeHalfLife.isEmpty() &&
           SourceStrengthUnits.isEmpty() &&
           ReferenceAirKermaRate.isEmpty() &&
           SourceStrength.isEmpty() &&
           SourceStrengthReferenceDate.isEmpty() &&
           SourceStrengthReferenceTime.isEmpty();
}

OFBool DRTSourceSequence::Item::isValid() const
{
    return !EmptyDefaultItem;
}

OFCondition DRTSourceSequence::Item::read(DcmItem &item)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        /* re-initialize object */
        clear();
        getAndCheckElementFromDataset(item, SourceNumber, "1", "1", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceSerialNumber, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceModelID, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceDescription, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceType, "1", "1", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceManufacturer, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, ActiveSourceDiameter, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, ActiveSourceLength, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, MaterialID, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceEncapsulationNominalThickness, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceEncapsulationNominalTransmission, "1", "3", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceIsotopeName, "1", "1", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceIsotopeHalfLife, "1", "1", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceStrengthUnits, "1", "1C", "SourceSequence");
        getAndCheckElementFromDataset(item, ReferenceAirKermaRate, "1", "1", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceStrength, "1", "1C", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceStrengthReferenceDate, "1", "1", "SourceSequence");
        getAndCheckElementFromDataset(item, SourceStrengthReferenceTime, "1", "1", "SourceSequence");
        result = EC_Normal;
    }
    return result;
}

OFCondition DRTSourceSequence::Item::write(DcmItem &item)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = EC_Normal;
        addElementToDataset(result, item, new DcmIntegerString(SourceNumber), "1", "1", "SourceSequence");
        addElementToDataset(result, item, new DcmLongString(SourceSerialNumber), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmShortString(SourceModelID), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmLongString(SourceDescription), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmCodeString(SourceType), "1", "1", "SourceSequence");
        addElementToDataset(result, item, new DcmLongString(SourceManufacturer), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmDecimalString(ActiveSourceDiameter), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmDecimalString(ActiveSourceLength), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmShortString(MaterialID), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmDecimalString(SourceEncapsulationNominalThickness), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmDecimalString(SourceEncapsulationNominalTransmission), "1", "3", "SourceSequence");
        addElementToDataset(result, item, new DcmLongString(SourceIsotopeName), "1", "1", "SourceSequence");
        addElementToDataset(result, item, new DcmDecimalString(SourceIsotopeHalfLife), "1", "1", "SourceSequence");
        addElementToDataset(result, item, new DcmCodeString(SourceStrengthUnits), "1", "1C", "SourceSequence");
        addElementToDataset(result, item, new DcmDecimalString(ReferenceAirKermaRate), "1", "1", "SourceSequence");
        addElementToDataset(result, item, new DcmDecimalString(SourceStrength), "1", "1C", "SourceSequence");
        addElementToDataset(result, item, new DcmDate(SourceStrengthReferenceDate), "1", "1", "SourceSequence");
        addElementToDataset(result, item, new DcmTime(SourceStrengthReferenceTime), "1", "1", "SourceSequence");
    }
    return result;
}

OFCondition DRTSourceSequence::Item::getActiveSourceDiameter(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(ActiveSourceDiameter, value, pos);
}

OFCondition DRTSourceSequence::Item::getActiveSourceDiameter(Float64 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmDecimalString &, ActiveSourceDiameter).getFloat64(value, pos);
}

OFCondition DRTSourceSequence::Item::getActiveSourceLength(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(ActiveSourceLength, value, pos);
}

OFCondition DRTSourceSequence::Item::getActiveSourceLength(Float64 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmDecimalString &, ActiveSourceLength).getFloat64(value, pos);
}

OFCondition DRTSourceSequence::Item::getMaterialID(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(MaterialID, value, pos);
}

OFCondition DRTSourceSequence::Item::getReferenceAirKermaRate(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(ReferenceAirKermaRate, value, pos);
}

OFCondition DRTSourceSequence::Item::getReferenceAirKermaRate(Float64 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmDecimalString &, ReferenceAirKermaRate).getFloat64(value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceDescription(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceDescription, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceEncapsulationNominalThickness(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceEncapsulationNominalThickness, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceEncapsulationNominalThickness(Float64 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmDecimalString &, SourceEncapsulationNominalThickness).getFloat64(value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceEncapsulationNominalTransmission(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceEncapsulationNominalTransmission, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceEncapsulationNominalTransmission(Float64 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmDecimalString &, SourceEncapsulationNominalTransmission).getFloat64(value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceIsotopeHalfLife(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceIsotopeHalfLife, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceIsotopeHalfLife(Float64 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmDecimalString &, SourceIsotopeHalfLife).getFloat64(value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceIsotopeName(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceIsotopeName, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceManufacturer(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceManufacturer, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceModelID(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceModelID, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceNumber(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceNumber, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceNumber(Sint32 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmIntegerString &, SourceNumber).getSint32(value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceSerialNumber(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceSerialNumber, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceStrength(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceStrength, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceStrength(Float64 &value, const unsigned long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return OFconst_cast(DcmDecimalString &, SourceStrength).getFloat64(value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceStrengthReferenceDate(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceStrengthReferenceDate, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceStrengthReferenceTime(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceStrengthReferenceTime, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceStrengthUnits(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceStrengthUnits, value, pos);
}

OFCondition DRTSourceSequence::Item::getSourceType(OFString &value, const signed long pos) const
{
    if (EmptyDefaultItem)
        return EC_IllegalCall;
    else
        return getStringValueFromElement(SourceType, value, pos);
}

OFCondition DRTSourceSequence::Item::setActiveSourceDiameter(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDecimalString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = ActiveSourceDiameter.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setActiveSourceLength(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDecimalString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = ActiveSourceLength.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setMaterialID(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmShortString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = MaterialID.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setReferenceAirKermaRate(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDecimalString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = ReferenceAirKermaRate.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceDescription(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmLongString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceDescription.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceEncapsulationNominalThickness(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDecimalString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceEncapsulationNominalThickness.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceEncapsulationNominalTransmission(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDecimalString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceEncapsulationNominalTransmission.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceIsotopeHalfLife(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDecimalString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceIsotopeHalfLife.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceIsotopeName(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmLongString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceIsotopeName.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceManufacturer(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmLongString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceManufacturer.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceModelID(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmShortString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceModelID.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceNumber(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmIntegerString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceNumber.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceSerialNumber(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmLongString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceSerialNumber.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceStrength(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDecimalString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceStrength.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceStrengthReferenceDate(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmDate::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceStrengthReferenceDate.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceStrengthReferenceTime(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmTime::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceStrengthReferenceTime.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceStrengthUnits(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmCodeString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceStrengthUnits.putOFStringArray(value);
    }
    return result;
}

OFCondition DRTSourceSequence::Item::setSourceType(const OFString &value, const OFBool check)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultItem)
    {
        result = (check) ? DcmCodeString::checkStringValue(value, "1") : EC_Normal;
        if (result.good())
            result = SourceType.putOFStringArray(value);
    }
    return result;
}

// --- sequence class ---

DRTSourceSequence::DRTSourceSequence(const OFBool emptyDefaultSequence)
  : EmptyDefaultSequence(emptyDefaultSequence),
    SequenceOfItems(),
    CurrentItem(),
    EmptyItem(OFTrue /*emptyDefaultItem*/)
{
    CurrentItem = SequenceOfItems.end();
}

DRTSourceSequence::DRTSourceSequence(const DRTSourceSequence &copy)
  : EmptyDefaultSequence(copy.EmptyDefaultSequence),
    SequenceOfItems(),
    CurrentItem(),
    EmptyItem(OFTrue /*emptyDefaultItem*/)
{
    /* create a copy of the internal sequence of items */
    Item *item = NULL;
    OFListConstIterator(Item *) current = copy.SequenceOfItems.begin();
    const OFListConstIterator(Item *) last = copy.SequenceOfItems.end();
    while (current != last)
    {
        item = new Item(**current);
        if (item != NULL)
        {
            SequenceOfItems.push_back(item);
        } else {
            /* memory exhausted, there is nothing we can do about it */
            break;
        }
        ++current;
    }
    CurrentItem = SequenceOfItems.begin();
}

DRTSourceSequence &DRTSourceSequence::operator=(const DRTSourceSequence &copy)
{
    if (this != &copy)
    {
        clear();
        EmptyDefaultSequence = copy.EmptyDefaultSequence;
        /* create a copy of the internal sequence of items */
        Item *item = NULL;
        OFListConstIterator(Item *) current = copy.SequenceOfItems.begin();
        const OFListConstIterator(Item *) last = copy.SequenceOfItems.end();
        while (current != last)
        {
            item = new Item(**current);
            if (item != NULL)
            {
                SequenceOfItems.push_back(item);
            } else {
                /* memory exhausted, there is nothing we can do about it */
                break;
            }
            ++current;
        }
        CurrentItem = SequenceOfItems.begin();
    }
    return *this;
}

DRTSourceSequence::~DRTSourceSequence()
{
    clear();
}

void DRTSourceSequence::clear()
{
    if (!EmptyDefaultSequence)
    {
        CurrentItem = SequenceOfItems.begin();
        const OFListConstIterator(Item *) last = SequenceOfItems.end();
        /* delete all items and free memory */
        while (CurrentItem != last)
        {
            delete (*CurrentItem);
            CurrentItem = SequenceOfItems.erase(CurrentItem);
        }
        /* make sure that the list is empty */
        SequenceOfItems.clear();
        CurrentItem = SequenceOfItems.end();
    }
}

OFBool DRTSourceSequence::isEmpty()
{
    return SequenceOfItems.empty();
}

OFBool DRTSourceSequence::isValid() const
{
    return !EmptyDefaultSequence;
}

unsigned long DRTSourceSequence::getNumberOfItems() const
{
    return SequenceOfItems.size();
}

OFCondition DRTSourceSequence::gotoFirstItem()
{
    OFCondition result = EC_IllegalCall;
    if (!SequenceOfItems.empty())
    {
        CurrentItem = SequenceOfItems.begin();
        result = EC_Normal;
    }
    return result;
}

OFCondition DRTSourceSequence::gotoNextItem()
{
    OFCondition result = EC_IllegalCall;
    if (CurrentItem != SequenceOfItems.end())
    {
        ++CurrentItem;
        result = EC_Normal;
    }
    return result;
}

OFCondition DRTSourceSequence::gotoItem(const unsigned long num, OFListIterator(Item *) &iterator)
{
    OFCondition result = EC_IllegalCall;
    if (!SequenceOfItems.empty())
    {
        unsigned long idx = num + 1;
        iterator = SequenceOfItems.begin();
        const OFListConstIterator(Item *) last = SequenceOfItems.end();
        while ((--idx > 0) && (iterator != last))
            ++iterator;
        /* specified list item found? */
        if ((idx == 0) && (iterator != last))
            result = EC_Normal;
        else
            result = EC_IllegalParameter;
    }
    return result;
}

OFCondition DRTSourceSequence::gotoItem(const unsigned long num, OFListConstIterator(Item *) &iterator) const
{
    OFCondition result = EC_IllegalCall;
    if (!SequenceOfItems.empty())
    {
        unsigned long idx = num + 1;
        iterator = SequenceOfItems.begin();
        const OFListConstIterator(Item *) last = SequenceOfItems.end();
        while ((--idx > 0) && (iterator != last))
            ++iterator;
        /* specified list item found? */
        if ((idx == 0) && (iterator != last))
            result = EC_Normal;
        else
            result = EC_IllegalParameter;
    }
    return result;
}

OFCondition DRTSourceSequence::gotoItem(const unsigned long num)
{
    return gotoItem(num, CurrentItem);
}

OFCondition DRTSourceSequence::getCurrentItem(Item *&item) const
{
    OFCondition result = EC_IllegalCall;
    if (CurrentItem != SequenceOfItems.end())
    {
        item = *CurrentItem;
        result = EC_Normal;
    }
    return result;
}

DRTSourceSequence::Item &DRTSourceSequence::getCurrentItem()
{
    if (CurrentItem != SequenceOfItems.end())
        return **CurrentItem;
    else
        return EmptyItem;
}

const DRTSourceSequence::Item &DRTSourceSequence::getCurrentItem() const
{
    if (CurrentItem != SequenceOfItems.end())
        return **CurrentItem;
    else
        return EmptyItem;
}

OFCondition DRTSourceSequence::getItem(const unsigned long num, Item *&item)
{
    OFListIterator(Item *) iterator;
    OFCondition result = gotoItem(num, iterator);
    if (result.good())
        item = *iterator;
    return result;
}

DRTSourceSequence::Item &DRTSourceSequence::getItem(const unsigned long num)
{
    OFListIterator(Item *) iterator;
    if (gotoItem(num, iterator).good())
        return **iterator;
    else
        return EmptyItem;
}

const DRTSourceSequence::Item &DRTSourceSequence::getItem(const unsigned long num) const
{
    OFListConstIterator(Item *) iterator;
    if (gotoItem(num, iterator).good())
        return **iterator;
    else
        return EmptyItem;
}

DRTSourceSequence::Item &DRTSourceSequence::operator[](const unsigned long num)
{
    return getItem(num);
}

const DRTSourceSequence::Item &DRTSourceSequence::operator[](const unsigned long num) const
{
    return getItem(num);
}

OFCondition DRTSourceSequence::addItem(Item *&item)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultSequence)
    {
        item = new Item();
        if (item != NULL)
        {
            SequenceOfItems.push_back(item);
            result = EC_Normal;
        } else
            result = EC_MemoryExhausted;
    }
    return result;
}

OFCondition DRTSourceSequence::insertItem(const unsigned long pos, Item *&item)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultSequence)
    {
        OFListIterator(Item *) iterator;
        result = gotoItem(pos, iterator);
        if (result.good())
        {
            item = new Item();
            if (item != NULL)
            {
                SequenceOfItems.insert(iterator, 1, item);
                result = EC_Normal;
            } else
                result = EC_MemoryExhausted;
        } else
            result = addItem(item);
    }
    return result;
}

OFCondition DRTSourceSequence::removeItem(const unsigned long pos)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultSequence)
    {
        OFListIterator(Item *) iterator;
        if (gotoItem(pos, iterator).good())
        {
            delete *iterator;
            iterator = SequenceOfItems.erase(iterator);
            result = EC_Normal;
        } else
            result = EC_IllegalParameter;
    }
    return result;
}

OFCondition DRTSourceSequence::read(DcmItem &dataset,
                                    const OFString &card,
                                    const OFString &type,
                                    const char *moduleName)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultSequence)
    {
        /* re-initialize object */
        clear();
        /* retrieve sequence element from dataset */
        DcmSequenceOfItems *sequence;
        result = dataset.findAndGetSequence(DCM_SourceSequence, sequence);
        if (sequence != NULL)
        {
            if (checkElementValue(*sequence, card, type, result, moduleName))
            {
                DcmStack stack;
                OFBool first = OFTrue;
                /* iterate over all sequence items */
                while (result.good() && sequence->nextObject(stack, first /*intoSub*/).good())
                {
                    DcmItem *ditem = OFstatic_cast(DcmItem *, stack.top());
                    if (ditem != NULL)
                    {
                        Item *item = new Item();
                        if (item != NULL)
                        {
                            result = item->read(*ditem);
                            if (result.good())
                            {
                                /* append new item to the end of the list */
                                SequenceOfItems.push_back(item);
                                first = OFFalse;
                            }
                        } else
                            result = EC_MemoryExhausted;
                    } else
                        result = EC_CorruptedData;
                }
            }
        } else {
            DcmSequenceOfItems element(DCM_SourceSequence);
            checkElementValue(element, card, type, result, moduleName);
        }
    }
    return result;
}

OFCondition DRTSourceSequence::write(DcmItem &dataset,
                                     const OFString &card,
                                     const OFString &type,
                                     const char *moduleName)
{
    OFCondition result = EC_IllegalCall;
    if (!EmptyDefaultSequence)
    {
        result = EC_MemoryExhausted;
        DcmSequenceOfItems *sequence = new DcmSequenceOfItems(DCM_SourceSequence);
        if (sequence != NULL)
        {
            result = EC_Normal;
            /* an empty optional sequence is not written */
            if ((type == "2") || !SequenceOfItems.empty())
            {
                OFListIterator(Item *) iterator = SequenceOfItems.begin();
                const OFListConstIterator(Item *) last = SequenceOfItems.end();
                /* iterate over all sequence items */
                while (result.good() && (iterator != last))
                {
                    DcmItem *item = new DcmItem();
                    if (item != NULL)
                    {
                        /* append new item to the end of the sequence */
                        result = sequence->append(item);
                        if (result.good())
                        {
                            result = (*iterator)->write(*item);
                            ++iterator;
                        } else
                            delete item;
                    } else
                        result = EC_MemoryExhausted;
                }
                if (result.good())
                {
                    /* insert sequence element into the dataset */
                    result = dataset.insert(sequence, OFTrue /*replaceOld*/);
                }
                if (DCM_dcmrtLogger.isEnabledFor(OFLogger::WARN_LOG_LEVEL))
                    checkElementValue(*sequence, card, type, result, moduleName);
                if (result.good())
                {
                    /* forget reference to sequence object (avoid deletion below) */
                    sequence = NULL;
                }
            }
            else if (type == "1")
            {
                /* empty type 1 sequence not allowed */
                result = RT_EC_InvalidValue;
                if (DCM_dcmrtLogger.isEnabledFor(OFLogger::WARN_LOG_LEVEL))
                    checkElementValue(*sequence, card, type, result, moduleName);
            }
            /* delete sequence (if not inserted into the dataset) */
            delete sequence;
        }
    }
    return result;
}

// end of source file

redpajama_set_name: RedPajamaGithub


text:
<!DOCTYPE html>

	<meta charset="utf-8">
	<meta name="viewport" content="width=device-width, initial-scale=1">
	<title>jQuery Mobile Collapsible Integration Test</title>

	<link rel="stylesheet"  href="../../../jquery.mobile-1.0.1.min.css" />
	<link rel="stylesheet" href="../../../external/qunit.css"/>

jQuery Mobile Collapsible Integration Test

Basic collapsible

Section A

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
				have the "collapsed" state; you need to expand the header to see me.

Section B

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
				have the "collapsed" state; you need to expand the header to see me.

Basic collapsible

Section A

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.

Section B

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.

Section C

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.

Basic collapsible

Section D

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.

Section E

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
				have the "collapsed" state; you need to expand the header to see me.

Themed collapsibles

Section A

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
				have the "collapsed" state; you need to expand the header to see me.

Section B

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
				have the "collapsed" state; you need to expand the header to see me.

Section B

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
				have the "collapsed" state; you need to expand the header to see me.

Themed collapsibles

Section A

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.

Section B

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.

Section C

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.

Section D

I'm the collapsible content in a set so this feels like an accordion. I'm hidden by default because I
					have the "collapsed" state; you need to expand the header to see me.


text:
<?php
namespace PHPVCSControl\ValueObjects;

interface RepositorySourceInterface
{

}


text:
using Stormancer.Core;
using Stormancer.Plugins.Chat;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Stormancer
{
    public static class ChatPluginExtensions
    {
        public static void AddChat(this ISceneHost scene )
        {
            scene.Metadata[ChatPlugin.METADATA_KEY] = "enabled";
        }
    }
}


text:
RSpec.describe RuboCop::Cop::Style::RedundantReturn, :config do
  subject(:cop) { described_class.new(config) }

  let(:cop_config) { { 'AllowMultipleReturnValues' => false } }

  it 'reports an offense for def with only a return' do
    expect_offense(<<-RUBY.strip_indent)
      def func
        return something
        ^^^^^^ Redundant `return` detected.
      ensure
        2
      end
    RUBY
  end

  it 'reports an offense for defs with only a return' do
    expect_offense(<<-RUBY.strip_indent)
      def Test.func
        return something
        ^^^^^^ Redundant `return` detected.
      end
    RUBY
  end

  it 'reports an offense for def ending with return' do
    expect_offense(<<-RUBY.strip_indent)
      def func
        one
        two
        return something
        ^^^^^^ Redundant `return` detected.
      end
    RUBY
  end

  it 'reports an offense for defs ending with return' do
    expect_offense(<<-RUBY.strip_indent)
      def self.func
        one
        two
        return something
        ^^^^^^ Redundant `return` detected.
      end
    RUBY
  end

  it 'accepts return in a non-final position' do
    expect_no_offenses(<<-RUBY.strip_indent)
      def func
        return something if something_else
      end
    RUBY
  end

  it 'does not blow up on empty method body' do
    expect_no_offenses(<<-RUBY.strip_indent)
      def func
      end
    RUBY
  end

  it 'does not blow up on empty if body' do
    expect_no_offenses(<<-RUBY.strip_indent)
      def func
        if x
        elsif y
        else
        end
      end
    RUBY
  end

  it 'auto-corrects by removing redundant returns' do
    src = <<-RUBY.strip_indent
      def func
        one
        two
        return something
      end
    RUBY
    result_src = <<-RUBY.strip_indent
      def func
        one
        two
        something
      end
    RUBY
    new_source = autocorrect_source(src)
    expect(new_source).to eq(result_src)
  end

  context 'when return has no arguments' do
    shared_examples 'common behavior' do |ret|
      let(:src) do
        <<-RUBY.strip_indent
          def func
            one
            two
            #{ret}
            # comment
          end
        RUBY
      end

      it "registers an offense for #{ret}" do
        inspect_source(src)
        expect(cop.offenses.size).to eq(1)
      end

      it "auto-corrects by replacing #{ret} with nil" do
        new_source = autocorrect_source(src)
        expect(new_source).to eq(<<-RUBY.strip_indent)
          def func
            one
            two
            nil
            # comment
          end
        RUBY
      end
    end

    it_behaves_like 'common behavior', 'return'
    it_behaves_like 'common behavior', 'return()'
  end

  context 'when multi-value returns are not allowed' do
    it 'reports an offense for def with only a return' do
      expect_offense(<<-RUBY.strip_indent)
        def func
          return something, test
          ^^^^^^ Redundant `return` detected. To return multiple values, use an array.
        end
      RUBY
    end

    it 'reports an offense for defs with only a return' do
      expect_offense(<<-RUBY.strip_indent)
        def Test.func
          return something, test
          ^^^^^^ Redundant `return` detected. To return multiple values, use an array.
        end
      RUBY
    end

    it 'reports an offense for def ending with return' do
      expect_offense(<<-RUBY.strip_indent)
        def func
          one
          two
          return something, test
          ^^^^^^ Redundant `return` detected. To return multiple values, use an array.
        end
      RUBY
    end

    it 'reports an offense for defs ending with return' do
      expect_offense(<<-RUBY.strip_indent)
        def self.func
          one
          two
          return something, test
          ^^^^^^ Redundant `return` detected. To return multiple values, use an array.
        end
      RUBY
    end

    it 'auto-corrects by making implicit arrays explicit' do
      src = <<-RUBY.strip_indent
        def func
          return  1, 2
        end
      RUBY
      # Just 1, 2 is not valid Ruby.
      result_src = <<-RUBY.strip_indent
        def func
          [1, 2]
        end
      RUBY
      new_source = autocorrect_source(src)
      expect(new_source).to eq(result_src)
    end

    it 'auto-corrects removes return when using an explicit hash' do
      src = <<-RUBY.strip_indent
        def func
          return {:a => 1, :b => 2}
        end
      RUBY
      # :a => 1, :b => 2 is not valid Ruby
      result_src = <<-RUBY.strip_indent
        def func
          {:a => 1, :b => 2}
        end
      RUBY
      new_source = autocorrect_source(src)
      expect(new_source).to eq(result_src)
    end

    it 'auto-corrects by making an implicit hash explicit' do
      src = <<-RUBY.strip_indent
        def func
          return :a => 1, :b => 2
        end
      RUBY
      # :a => 1, :b => 2 is not valid Ruby
      result_src = <<-RUBY.strip_indent
        def func
          {:a => 1, :b => 2}
        end
      RUBY
      new_source = autocorrect_source(src)
      expect(new_source).to eq(result_src)
    end
  end

  context 'when multi-value returns are allowed' do
    let(:cop_config) { { 'AllowMultipleReturnValues' => true } }

    it 'accepts def with only a return' do
      expect_no_offenses(<<-RUBY.strip_indent)
        def func
          return something, test
        end
      RUBY
    end

    it 'accepts defs with only a return' do
      expect_no_offenses(<<-RUBY.strip_indent)
        def Test.func
          return something, test
        end
      RUBY
    end

    it 'accepts def ending with return' do
      expect_no_offenses(<<-RUBY.strip_indent)
        def func
          one
          two
          return something, test
        end
      RUBY
    end

    it 'accepts defs ending with return' do
      expect_no_offenses(<<-RUBY.strip_indent)
        def self.func
          one
          two
          return something, test
        end
      RUBY
    end

    it 'does not auto-correct' do
      src = <<-RUBY.strip_indent
        def func
          return  1, 2
        end
      RUBY
      new_source = autocorrect_source(src)
      expect(new_source).to eq(src)
    end
  end

  context 'when return is inside begin-end body' do
    let(:src) do
      <<-RUBY.strip_indent
        def func
          begin
            return 1
          end
        end
      RUBY
    end

    it 'registers an offense' do
      expect_offense(<<-RUBY.strip_indent)
        def func
          begin
            return 1
            ^^^^^^ Redundant `return` detected.
          end
        end
      RUBY
    end

    it 'auto-corrects' do
      corrected = autocorrect_source(src)
      expect(corrected).to eq <<-RUBY.strip_indent
        def func
          begin
            1
          end
        end
      RUBY
    end
  end

  context 'when rescue and return blocks present' do
    let(:src) do
      <<-RUBY.strip_indent
        def func
          1
          2
          return 3
        rescue SomeException
          4
          return 5
        rescue AnotherException
          return 6
        ensure
          return 7
        end
      RUBY
    end

    it 'does register an offense when inside function or rescue block' do
      expect_offense(<<-RUBY.strip_indent)
        def func
          1
          2
          return 3
          ^^^^^^ Redundant `return` detected.
        rescue SomeException
          4
          return 5
          ^^^^^^ Redundant `return` detected.
        rescue AnotherException
          return 6
          ^^^^^^ Redundant `return` detected.
        ensure
          return 7
        end
      RUBY
    end

    it 'auto-corrects' do
      corrected = autocorrect_source(src)
      expect(corrected).to eq <<-RUBY.strip_indent
        def func
          1
          2
          3
        rescue SomeException
          4
          5
        rescue AnotherException
          6
        ensure
          return 7
        end
      RUBY
    end
  end

  context 'when return is inside an if-branch' do
    let(:src) do
      <<-RUBY.strip_indent
        def func
          if x
            return 1
          elsif y
            return 2
          else
            return 3
          end
        end
      RUBY
    end

    it 'registers an offense' do
      expect_offense(<<-RUBY.strip_indent)
        def func
          if x
            return 1
            ^^^^^^ Redundant `return` detected.
          elsif y
            return 2
            ^^^^^^ Redundant `return` detected.
          else
            return 3
            ^^^^^^ Redundant `return` detected.
          end
        end
      RUBY
    end

    it 'auto-corrects' do
      corrected = autocorrect_source(src)
      expect(corrected).to eq <<-RUBY.strip_indent
        def func
          if x
            1
          elsif y
            2
          else
            3
          end
        end
      RUBY
    end
  end

  context 'when return is inside a when-branch' do
    let(:src) do
      <<-RUBY.strip_indent
        def func
          case x
          when y then return 1
          when z then return 2
          when q
          else
            return 3
          end
        end
      RUBY
    end

    it 'registers an offense' do
      expect_offense(<<-RUBY.strip_indent)
        def func
          case x
          when y then return 1
                      ^^^^^^ Redundant `return` detected.
          when z then return 2
                      ^^^^^^ Redundant `return` detected.
          when q
          else
            return 3
            ^^^^^^ Redundant `return` detected.
          end
        end
      RUBY
    end

    it 'auto-corrects' do
      corrected = autocorrect_source(src)
      expect(corrected).to eq <<-RUBY.strip_indent
        def func
          case x
          when y then 1
          when z then 2
          when q
          else
            3
          end
        end
      RUBY
    end
  end

  context 'when case nodes are empty' do
    it 'accepts empty when nodes' do
      expect_no_offenses(<<-RUBY.strip_indent)
        def func
          case x
          when y then 1
          when z # do nothing
          else
            3
          end
        end
      RUBY
    end
  end
end


text:
package com.intellij.xdebugger.impl.frame.actions;

import com.intellij.openapi.actionSystem.AnAction;
import com.intellij.openapi.actionSystem.AnActionEvent;
import com.intellij.xdebugger.impl.ui.tree.XDebuggerTree;
import org.jetbrains.annotations.NotNull;

import javax.swing.tree.TreeNode;
import javax.swing.tree.TreePath;
import java.util.ArrayList;
import java.util.List;

/**
 * @author nik
 */
public abstract class XWatchesTreeActionBase extends AnAction {
  protected static <T extends TreeNode> List<? extends T> getSelectedNodes(final @NotNull XDebuggerTree tree, Class<T> nodeClass) {
    List<T> list = new ArrayList<T>();
    TreePath[] selectionPaths = tree.getSelectionPaths();
    if (selectionPaths != null) {
      for (TreePath selectionPath : selectionPaths) {
        Object element = selectionPath.getLastPathComponent();
        if (nodeClass.isInstance(element)) {
          list.add(nodeClass.cast(element));
        }
      }
    }
    return list;
  }

  public void update(final AnActionEvent e) {
    boolean enabled = isEnabled(e);
    e.getPresentation().setEnabled(enabled);
  }

  protected abstract boolean isEnabled(AnActionEvent e);
}