def test_job_creation_after_pipegraph_run_raises(self):
        def inner():
            ppg.FileGeneratingJob("A", lambda: None)

        ppg.new_pipegraph(quiet=True, dump_graph=False)
        ppg.run_pipegraph()
        assertRaises(ValueError, inner)

        def test_plotjob_fails(self):
            def calc():
                return None

            def calc2():
                return pd.DataFrame(
                    {"X": list(range(0, 100)), "Y": list(range(50, 150)), "w": "B"}
                )

            def plot(df):
                return pyggplot.Plot(df).add_scatter("X", "Y")

            p1 = ppg.PlotJob("out/A.png", calc, plot)
            p2 = ppg.PlotJob("out/B.png", calc2, plot)
            import pathlib

            pc = ppg.CombinedPlotJob(
                pathlib.Path("out/C.png"), [p1, p2], {"facet": "w"}
            )
            with pytest.raises(ValueError):
                ppg.CombinedPlotJob(pathlib.Path("out/C.png"), [p1, p2], [])
            with pytest.raises(ValueError):
                ppg.CombinedPlotJob(pathlib.Path("out/C.png"), [p1], {"facet": "w"})

            ppg.CombinedPlotJob(pathlib.Path("out/D.png"), [p1, p2], [])
            ppg.CombinedPlotJob(pathlib.Path("out/E.png"), [p1, p2], {"facet": "w"})

            with pytest.raises(ppg.RuntimeError):
                ppg.run_pipegraph()
            assert "did not return a" in str(p1.cache_job.exception)
            assert pc.error_reason == "Indirect"

    def test_unpickle_bug_prevents_single_job_from_unpickling(self):
        def do_a():
            write("out/A", "A")
            append("out/As", "A")

        ppg.FileGeneratingJob("out/A", do_a)

        def do_b():
            write("out/B", "A")
            append("out/Bs", "A")

        job_B = ppg.FileGeneratingJob("out/B", do_b)
        cd = CantDepickle()
        job_parameter_unpickle_problem = ppg.ParameterInvariant("C", (cd,))
        job_B.depends_on(job_parameter_unpickle_problem)
        ppg.run_pipegraph()
        assert read("out/A") == "A"
        assert read("out/As") == "A"
        assert read("out/B") == "A"
        assert read("out/Bs") == "A"
        print("second run")
        ppg.new_pipegraph(dump_graph=False)

        ppg.FileGeneratingJob("out/A", do_a)
        job_B = ppg.FileGeneratingJob("out/B", do_b)
        job_parameter_unpickle_problem = ppg.ParameterInvariant("C", (cd,))
        job_B.depends_on(job_parameter_unpickle_problem)
        with pytest.raises(ppg.RuntimeError):
            ppg.run_pipegraph()
        assert read("out/A") == "A"
        assert read("out/As") == "A"
        assert read("out/B") == "A"
        assert (
            read("out/Bs") == "AA"
        )  # this one got rerun because we could not load the invariant...

    def test_filegen_invalidated_jobgen_created_filegen_later_also_invalidated(
        self, new_pipegraph
    ):
        a = ppg.FileGeneratingJob("out/A", lambda: writeappend("out/A", "out/Ac", "A"))
        p = ppg.ParameterInvariant("p", "p")
        a.depends_on(p)

        def gen():
            c = ppg.FileGeneratingJob(
                "out/C", lambda: writeappend("out/C", "out/Cx", "C")
            )
            c.depends_on(a)

        ppg.JobGeneratingJob("b", gen)
        ppg.run_pipegraph()
        assert read("out/A") == "A"
        assert read("out/Ac") == "A"
        assert read("out/C") == "C"
        assert read("out/Cx") == "C"
        new_pipegraph.new_pipegraph()

        a = ppg.FileGeneratingJob("out/A", lambda: writeappend("out/A", "out/Ac", "A"))
        p = ppg.ParameterInvariant("p", "p2")
        a.depends_on(p)
        ppg.JobGeneratingJob("b", gen)
        ppg.run_pipegraph()
        assert read("out/Ac") == "AA"
        assert read("out/Cx") == "CC"

 def test_jobs_concurrent_jobs_run_concurrently(self):
     # we'll determine this by the start respective end times..
     ppg.new_pipegraph(
         ppg.resource_coordinators.LocalSystem(max_cores_to_use=2),
         quiet=True,
         dump_graph=False,
     )
     jobA = ppg.FileGeneratingJob("out/A", lambda: write("out/A", "A"))
     jobB = ppg.FileGeneratingJob("out/B", lambda: write("out/B", "B"))
     jobA.cores_needed = 1
     jobB.cores_needed = 1
     ppg.run_pipegraph()
     assert read("out/A") == "A"
     assert read("out/B") == "B"
     if jobA.start_time < jobB.start_time:
         first_job = jobA
         second_job = jobB
     else:
         first_job = jobB
         second_job = jobA
     print(
         "times",
         first_job.start_time,
         first_job.stop_time,
         second_job.start_time,
         second_job.stop_time,
     )
     if jobA.start_time is None:
         raise ValueError("JobA did not run")
     assert first_job.stop_time > second_job.start_time

        def test_reruns_just_plot_if_plot_changed(self, new_pipegraph):
            def calc():
                append("out/calc", "A")
                return pd.DataFrame(
                    {"X": list(range(0, 100)), "Y": list(range(50, 150))}
                )

            def plot(df):
                append("out/plot", "B")
                return pyggplot.Plot(df).add_scatter("X", "Y")

            of = "out/test.png"
            ppg.PlotJob(of, calc, plot)
            ppg.run_pipegraph()
            assert magic(of).find(b"PNG image") != -1
            assert read("out/calc") == "A"
            assert read("out/plot") == "B"

            new_pipegraph.new_pipegraph()

            def plot2(df):
                append("out/plot", "B")
                return pyggplot.Plot(df).add_scatter("Y", "X")

            ppg.PlotJob(of, calc, plot2)
            ppg.run_pipegraph()
            assert magic(of).find(b"PNG image") != -1
            assert read("out/calc") == "A"
            assert read("out/plot") == "BB"

        def test_no_rerun_if_calc_change_but_ignore_codechanges(self, new_pipegraph):
            def calc():
                append("out/calc", "A")
                return pd.DataFrame(
                    {"X": list(range(0, 100)), "Y": list(range(50, 150))}
                )

            def plot(df):
                append("out/plot", "B")
                return pyggplot.Plot(df).add_scatter("X", "Y")

            of = "out/test.png"
            job = ppg.PlotJob(of, calc, plot)
            ppg.run_pipegraph()
            assert magic(of).find(b"PNG image") != -1
            assert read("out/calc") == "A"
            assert read("out/plot") == "B"

            new_pipegraph.new_pipegraph()

            def calc2():
                append("out/calc", "A")
                x = 5  # noqa: E157,F841
                return pd.DataFrame(
                    {"X": list(range(0, 100)), "Y": list(range(50, 150))}
                )

            job = ppg.PlotJob(of, calc2, plot)
            job.ignore_code_changes()
            ppg.run_pipegraph()
            assert magic(of).find(b"PNG image") != -1
            assert read("out/calc") == "A"
            assert read("out/plot") == "B"

    def test_no_rerun_if_ignore_code_changes_and_plot_changes(self):
        import pydataframe
        def calc():
            append('out/calc', 'A')
            return pydataframe.DataFrame({"X": list(range(0, 100)), 'Y': list(range(50, 150))})
        def plot(df):
            append('out/plot', 'B')
            return pyggplot.Plot(df).add_scatter('X','Y')
        of = 'out/test.png'
        job = ppg.PlotJob(of, calc, plot)
        ppg.run_pipegraph()
        self.assertTrue(magic(of).find('PNG image') != -1)
        self.assertEqual(read('out/calc'),'A')
        self.assertEqual(read('out/plot'),'B')

        ppg.new_pipegraph(rc_gen(), quiet=True)
        def plot2(df):
            append('out/plot', 'B')
            return pyggplot.Plot(df).add_scatter('Y','X')
        job = ppg.PlotJob(of, calc, plot2)
        job.ignore_code_changes()
        ppg.run_pipegraph()
        self.assertTrue(magic(of).find('PNG image') != -1)
        self.assertEqual(read('out/calc'),'A')
        self.assertEqual(read('out/plot'),'B')

    def test_raises_on_non_dependend_job_injection2(self):
        o = Dummy()
        of = "out/A"

        def do_write():
            write(of, o.A + o.B)

        job = ppg.FileGeneratingJob(of, do_write)
        ppg.FileGeneratingJob("out/D", lambda: write("out/D", "D"))

        def generate_deps():
            def load_a():
                return "A"

            def load_b():
                return "B"

            dlA = ppg.AttributeLoadingJob("dlA", o, "A", load_a)
            ppg.AttributeLoadingJob("dlB", o, "B", load_b)
            job.depends_on(dlA)
            # let's not do anything with dlA

        gen_job = ppg.DependencyInjectionJob("C", generate_deps)
        job.depends_on(gen_job)
        with pytest.raises(ppg.RuntimeError):
            ppg.run_pipegraph()

        assert not (os.path.exists(of))  # since the gen job crashed
        assert os.path.exists(
            "out/D"
        )  # since it has no relation to the gen job actually...
        assert isinstance(gen_job.exception, ppg.JobContractError)
        assert "case 1" in str(gen_job.exception)

    def test_raises_on_non_dependend_job_injection2_can_be_ignored(self):
        o = Dummy()
        of = "out/A"

        def do_write():
            write(of, o.A)  # + o.B - but B is not in the dependency chain!

        job = ppg.FileGeneratingJob(of, do_write)
        ppg.FileGeneratingJob("out/D", lambda: write("out/D", "D"))

        def generate_deps():
            def load_a():
                return "A"

            def load_b():
                return "B"

            dlA = ppg.AttributeLoadingJob("dlA", o, "A", load_a)
            ppg.AttributeLoadingJob("dlB", o, "B", load_b)
            job.depends_on(dlA)
            # let's not do anything with dlA

        gen_job = ppg.DependencyInjectionJob(
            "C", generate_deps, check_for_dependency_injections=False
        )
        job.depends_on(gen_job)
        ppg.run_pipegraph()

        assert os.path.exists(of)  # since the gen job crashed

    def test_exceeding_max_cycle(self, new_pipegraph):
        max_depth = 50
        # this raisess...
        jobs = []
        for x in range(0, max_depth - 1):
            j = ppg.FileGeneratingJob(str(x), lambda: write(str(x), str(x)))
            if jobs:
                j.depends_on(jobs[-1])
            jobs.append(j)
        jobs[0].depends_on(j)

        def inner():
            ppg.run_pipegraph()

        assertRaises(ppg.CycleError, inner)

        new_pipegraph.new_pipegraph()
        jobs = []
        for x in range(0, max_depth + 100):
            j = ppg.FileGeneratingJob(str(x), lambda: write(str(x), str(x)))
            if jobs:
                j.depends_on(jobs[-1])
            jobs.append(j)
        jobs[0].depends_on(j)

        with pytest.raises(ppg.CycleError):
            ppg.run_pipegraph()

    def test_reruns_both_if_calc_changed(self):
        import pydataframe
        def calc():
            append('out/calc', 'A')
            return pydataframe.DataFrame({"X": list(range(0, 100)), 'Y': list(range(50, 150))})
        def plot(df):
            append('out/plot', 'B')
            return pyggplot.Plot(df).add_scatter('X','Y')
        of = 'out/test.png'
        job = ppg.PlotJob(of, calc, plot)
        ppg.run_pipegraph()
        self.assertTrue(magic(of).find('PNG image') != -1)
        self.assertEqual(read('out/calc'),'A')
        self.assertEqual(read('out/plot'),'B')

        ppg.new_pipegraph(rc_gen(), quiet=True)
        def calc2():
            append('out/calc', 'A')
            x = 5
            return pydataframe.DataFrame({"X": list(range(0, 100)), 'Y': list(range(50, 150))})
        job = ppg.PlotJob(of, calc2, plot)
        ppg.run_pipegraph()
        self.assertTrue(magic(of).find('PNG image') != -1)
        self.assertEqual(read('out/calc'),'AA')
        self.assertEqual(read('out/plot'),'BB')

 def test_basic_prune(self):
     ppg.FileGeneratingJob("A", lambda: write("A", "A"))
     b = ppg.FileGeneratingJob("B", lambda: write("B", "B"))
     b.prune()
     ppg.run_pipegraph()
     assert Path("A").read_text() == "A"
     assert not Path("B").exists()

        def test_basic(self):
            def calc():
                return pd.DataFrame(
                    {"X": list(range(0, 100)), "Y": list(range(50, 150))}
                )

            def plot(df):
                return pyggplot.Plot(df).add_scatter("X", "Y")

            def plot2(df):
                p = pyggplot.Plot(df).add_scatter("Y", "X")
                p.width = 5
                p.height = 2
                return p

            of = "out/test.png"
            p = ppg.PlotJob(of, calc, plot)
            p.add_fiddle(lambda p: p.scale_x_log10())
            p.add_another_plot("out/test2.png", plot2)
            ppg.run_pipegraph()
            assert magic(of).find(b"PNG image") != -1
            assert os.path.exists(of + ".tsv")
            assert os.path.exists("cache/out/test.png")
            assert os.path.exists("out/test2.png")
            assert not os.path.exists("cache/out/test2.png")
            assert not os.path.exists("cache/out/test2.png.tsv")

 def test_tempfile_not_run_on_prune(self):
      a = ppg.TempFileGeneratingJob("A", lambda: write("A", "A"))
      b = ppg.FileGeneratingJob("B", lambda: write("B", "B" + read("A")))
      b.depends_on(a)
      b.prune()
      ppg.run_pipegraph()
      assert not Path('B').exists()
      assert not Path('A').exists()

    def test_run_may_be_called_only_once(self):
        ppg.new_pipegraph(quiet=True, dump_graph=False)
        ppg.run_pipegraph()

        def inner():
            ppg.run_pipegraph()

        assertRaises(ValueError, inner)

 def test_pruning_final_jobs_directly(self):
     ppg.FileGeneratingJob("A", lambda: write("A", "A"))
     ppg.FileGeneratingJob("B", lambda: write("B", "B"))
     c = ppg.FinalJob("shu", lambda: write("C", "C"))
     c.prune()
     ppg.run_pipegraph()
     assert Path("A").read_text() == "A"
     assert Path("B").read_text() == "B"
     assert not Path("C").exists()

    def test_can_not_run_twice(self):

        ppg.new_pipegraph(dump_graph=False)
        ppg.run_pipegraph()
        try:
            ppg.run_pipegraph()
            assert False  # "Exception not correctly raised"
        except ValueError as e:
            print(e)
            assert "Each pipegraph may be run only once." in str(e)

 def test_pdf(self):
     import pydataframe
     def calc():
         return pydataframe.DataFrame({"X": list(range(0, 100)), 'Y': list(range(50, 150))})
     def plot(df):
         return pyggplot.Plot(df).add_scatter('X','Y')
     of = 'out/test.pdf'
     job = ppg.PlotJob(of, calc, plot)
     ppg.run_pipegraph()
     self.assertTrue(magic(of).find('PDF document') != -1)

 def test_basic(self):
     ppg.new_pipegraph(rc_gen(), quiet=False)
     import pydataframe
     def calc():
         return pydataframe.DataFrame({"X": list(range(0, 100)), 'Y': list(range(50, 150))})
     def plot(df):
         return pyggplot.Plot(df).add_scatter('X','Y')
     of = 'out/test.png'
     job = ppg.PlotJob(of, calc, plot)
     ppg.run_pipegraph()
     self.assertTrue(magic(of).find('PNG image') != -1)