def test_running_interactive_command(self):
        sourcetree = SourceTree()

        command = "python3 -c \"print('input please?'); a = input();print('OK' if a=='yes' else 'NO')\""
        output = sourcetree.run_command(command, user_input='no')
        assert 'NO' in output
        output = sourcetree.run_command(command, user_input='yes')
        assert 'OK' in output

    def test_running_interactive_command(self):
        sourcetree = SourceTree()
        sourcetree.run_command("mkdir superlists", cwd=sourcetree.tempdir)

        command = "python3 -c \"print('input please?'); a = input();print('OK' if a=='yes' else 'NO')\""
        output = sourcetree.run_command(command, user_input="no")
        assert "NO" in output
        output = sourcetree.run_command(command, user_input="yes")
        assert "OK" in output

 def test_special_cases_fab_deploy(self, mock_subprocess):
     mock_subprocess.Popen.return_value.returncode = 0
     mock_subprocess.Popen.return_value.communicate.return_value = 'a', 'b'
     sourcetree = SourceTree()
     sourcetree.run_command('fab deploy:[email protected]')
     expected = (
         'cd deploy_tools &&'
         ' fab -D -i'
         ' ~/Dropbox/Book/.vagrant/machines/default/virtualbox/private_key'
         ' deploy:[email protected]'
     )
     assert mock_subprocess.Popen.call_args[0][0] == expected

 def setUp(self):
     self.sourcetree = SourceTree()
     self.tempdir = self.sourcetree.tempdir
     self.processes = []
     self.pos = 0
     self.dev_server_running = False
     self.current_server_cd = None

 def setUp(self):
     self.sourcetree = SourceTree()
     self.sourcetree.get_local_repo_path = lambda c: os.path.abspath(os.path.join(
         os.path.dirname(__file__), 'testrepo'
     ))
     self.sourcetree.start_with_checkout(17)
     self.sourcetree.run_command('git checkout test-start')
     self.sourcetree.run_command('git reset')

 def test_checks_out_repo_current_chapter_as_master(self):
     sourcetree = SourceTree()
     sourcetree.get_local_repo_path = lambda c: os.path.abspath(os.path.join(os.path.dirname(__file__), "testrepo"))
     sourcetree.start_with_checkout(21)
     remotes = sourcetree.run_command("git remote").split()
     assert remotes == ["repo"]
     branch = sourcetree.run_command("git branch").strip()
     assert branch == "* master"
     diff = sourcetree.run_command("git diff repo/chapter_20").strip()
     assert diff == ""

 def test_checks_out_repo_chapter_as_master(self):
     sourcetree = SourceTree()
     sourcetree.get_local_repo_path = lambda c: os.path.abspath(os.path.join(
         os.path.dirname(__file__), 'testrepo'
     ))
     sourcetree.start_with_checkout('chapter_17', 'chapter_16')
     remotes = sourcetree.run_command('git remote').split()
     assert remotes == ['repo']
     branch = sourcetree.run_command('git branch').strip()
     assert branch == '* master'
     diff = sourcetree.run_command('git diff repo/chapter_16').strip()
     assert diff == ''

    def test_cleanup_kills_backgrounded_processes_and_rmdirs(self):
        sourcetree = SourceTree()
        sourcetree.run_command('python -c"import time; time.sleep(5)" & #runserver', cwd=sourcetree.tempdir)
        assert len(sourcetree.processes) == 1
        sourcetree_pid = sourcetree.processes[0].pid
        pids = subprocess.check_output('pgrep -f time.sleep', shell=True).decode('utf8').split()
        print('sourcetree_pid', sourcetree_pid)
        print('pids', pids)
        sids = []
        for pid in reversed(pids):
            print('checking', pid)
            cmd = 'ps -o sid --no-header -p %s' % (pid,)
            print(cmd)
            try:
                sid = subprocess.check_output(cmd, shell=True)
                print('sid', sid)
                sids.append(sid)
                assert sourcetree_pid == int(sid)
            except subprocess.CalledProcessError:
                pass
        assert sids

        sourcetree.cleanup()
        assert 'time.sleep' not in subprocess.check_output('ps auxf', shell=True).decode('utf8')

 def test_special_cases_wget_bootstrap(self):
     sourcetree = SourceTree()
     sourcetree.run_command("mkdir superlists", cwd=sourcetree.tempdir)
     with patch("sourcetree.subprocess") as mock_subprocess:
         mock_subprocess.Popen.return_value.communicate.return_value = ("bla bla", None)
         sourcetree.run_command(BOOTSTRAP_WGET)
         assert not mock_subprocess.Popen.called
     assert os.path.exists(os.path.join(sourcetree.tempdir, "superlists", "bootstrap.zip"))
     diff = sourcetree.run_command(
         "diff %s bootstrap.zip" % (os.path.join(os.path.dirname(__file__), "..", "downloads", "bootstrap-3.0.zip"))
     )
     assert diff == ""

 def test_special_cases_wget_bootstrap(self):
     sourcetree = SourceTree()
     sourcetree.run_command('mkdir superlists', cwd=sourcetree.tempdir)
     with patch('sourcetree.subprocess') as mock_subprocess:
         mock_subprocess.Popen.return_value.communicate.return_value = (
                 'bla bla', None
         )
         sourcetree.run_command(BOOTSTRAP_WGET)
         assert not mock_subprocess.Popen.called
     assert os.path.exists(os.path.join(sourcetree.tempdir, 'superlists', 'bootstrap.zip'))
     diff = sourcetree.run_command('diff %s bootstrap.zip' % (
         os.path.join(os.path.dirname(__file__), '..', 'downloads', 'bootstrap.zip'))
     )
     assert diff == ''

 def test_doesnt_raise_for_some_things_where_a_return_code_is_ok(self):
     sourcetree = SourceTree()
     sourcetree.run_command('diff foo bar', cwd=sourcetree.tempdir)
     sourcetree.run_command('python test.py', cwd=sourcetree.tempdir)

 def test_environment_variables(self):
     sourcetree = SourceTree()
     os.environ['TEHFOO'] = 'baz'
     output = sourcetree.run_command('echo $TEHFOO', cwd=sourcetree.tempdir)
     assert output.strip() == 'baz'

 def test_raises_on_errors(self):
     sourcetree = SourceTree()
     with self.assertRaises(Exception):
         sourcetree.run_command('synt!tax error', cwd=sourcetree.tempdir)
     sourcetree.run_command('synt!tax error', cwd=sourcetree.tempdir, ignore_errors=True)

 def test_returns_output(self):
     sourcetree = SourceTree()
     output = sourcetree.run_command('echo hello', cwd=sourcetree.tempdir)
     assert output == 'hello\n'

 def test_running_simple_command(self):
     sourcetree = SourceTree()
     sourcetree.run_command('touch foo', cwd=sourcetree.tempdir)
     assert os.path.exists(os.path.join(sourcetree.tempdir, 'foo'))

 def test_get_contents(self):
     sourcetree = SourceTree()
     os.makedirs(sourcetree.tempdir + "/superlists")
     with open(sourcetree.tempdir + "/superlists/foo.txt", "w") as f:
         f.write("bla bla")
     assert sourcetree.get_contents("foo.txt") == "bla bla"

class ChapterTest(unittest.TestCase):
    maxDiff = None

    def setUp(self):
        self.sourcetree = SourceTree()
        self.tempdir = self.sourcetree.tempdir
        self.processes = []
        self.pos = 0
        self.dev_server_running = False
        self.current_server_cd = None


    def tearDown(self):
        self.sourcetree.cleanup()


    def parse_listings(self):
        base_dir = os.path.split(os.path.abspath(os.path.dirname(__file__)))[0]
        filename = 'chapter_{0:02d}.html'.format(self.chapter_no)
        with open(os.path.join(base_dir, filename), encoding='utf-8') as f:
            raw_html = f.read()
        parsed_html = html.fromstring(raw_html)
        listings_nodes = parsed_html.cssselect('div.listingblock')
        self.listings = [p for n in listings_nodes for p in parse_listing(n)]


    def check_final_diff(
        self, chapter, ignore_moves=False, ignore_secret_key=False,
        diff=None
    ):
        if diff is None:
            diff = self.run_command(Command(
                'git diff -b repo/chapter_{0:02d}'.format(chapter)
            ))
        try:
            print('checking final diff', diff)
        except io.BlockingIOError:
            pass
        self.assertNotIn('fatal:', diff)
        start_marker = 'diff --git a/\n'
        commit = Commit.from_diff(start_marker + diff)
        error = AssertionError('Final diff was not empty, was:\n{}'.format(diff))

        if ignore_secret_key:
            for line in commit.lines_to_add + commit.lines_to_remove:
                if 'SECRET_KEY' not in line:
                    raise error

        elif ignore_moves:
            if commit.deleted_lines or commit.new_lines:
                raise AssertionError(
                    'Found lines to delete or add in diff.\nto delete:\n{}\n\nto add:\n{}'.format(
                        '\n- '.join(commit.deleted_lines), '\n+'.join(commit.new_lines)
                    )
                )

        elif commit.lines_to_add or commit.lines_to_remove:
            raise error


    def write_to_file(self, codelisting):
        self.assertEqual(
            type(codelisting), CodeListing,
            "passed a non-Codelisting to write_to_file:\n%s" % (codelisting,)
        )
        print('writing to file', codelisting.filename)
        write_to_file(codelisting, os.path.join(self.tempdir, 'superlists'))
        filenames = codelisting.filename.split(', ')
        for filename in filenames:
            with open(os.path.join(self.tempdir, 'superlists', filename)) as f:
                print('wrote:')
                print(f.read())


    def apply_patch(self, codelisting):
        tf = tempfile.NamedTemporaryFile(delete=False)
        tf.write(codelisting.contents.encode('utf8'))
        tf.write('\n'.encode('utf8'))
        tf.close()
        print('patch:\n', codelisting.contents)
        patch_output = self.run_command(
            Command('patch --fuzz=3 --no-backup-if-mismatch %s %s' % (codelisting.filename, tf.name))
        )
        print(patch_output)
        self.assertNotIn('malformed', patch_output)
        self.assertNotIn('failed', patch_output.lower())
        codelisting.was_checked = True
        with open(os.path.join(self.tempdir, 'superlists', codelisting.filename)) as f:
            print(f.read())
        os.remove(tf.name)
        self.pos += 1
        codelisting.was_written = True


    def run_command(self, command, cwd=None, user_input=None, ignore_errors=False):
        self.assertEqual(
            type(command), Command,
            "passed a non-Command to run-command:\n%s" % (command,)
        )
        if command == 'git push':
#.........这里部分代码省略.........

 def test_default_directory_is_superlists(self):
     sourcetree = SourceTree()
     os.makedirs(os.path.join(sourcetree.tempdir, "superlists"))
     sourcetree.run_command("touch foo")
     assert os.path.exists(os.path.join(sourcetree.tempdir, "superlists", "foo"))

 def test_default_directory_is_tempdir(self):
     sourcetree = SourceTree()
     sourcetree.run_command('touch foo')
     assert os.path.exists(os.path.join(sourcetree.tempdir, 'foo'))

class ApplyFromGitRefTest(unittest.TestCase):

    def setUp(self):
        self.sourcetree = SourceTree()
        self.sourcetree.get_local_repo_path = lambda c: os.path.abspath(os.path.join(
            os.path.dirname(__file__), 'testrepo'
        ))
        self.sourcetree.start_with_checkout(17)
        self.sourcetree.run_command('git checkout test-start')
        self.sourcetree.run_command('git reset')


    def test_from_real_git_stuff(self):
        listing = CodeListing(filename='file1.txt', contents=dedent(
            """
            file 1 line 2 amended
            file 1 line 3
            """).lstrip()
        )
        listing.commit_ref = 'ch17l021'

        self.sourcetree.apply_listing_from_commit(listing)


        with open(self.sourcetree.tempdir + '/superlists/file1.txt') as f:
            assert f.read() == dedent(
                """
                file 1 line 1
                file 1 line 2 amended
                file 1 line 3
                """).lstrip()

        assert listing.was_written


    def test_leaves_staging_empty(self):
        listing = CodeListing(filename='file1.txt', contents=dedent(
            """
            file 1 line 2 amended
            file 1 line 3
            """).lstrip()
        )
        listing.commit_ref = 'ch17l021'

        self.sourcetree.apply_listing_from_commit(listing)

        staged = self.sourcetree.run_command('git diff --staged')
        assert staged == ''


    def test_raises_if_wrong_file(self):
        listing = CodeListing(filename='file2.txt', contents=dedent(
            """
            file 1 line 1
            file 1 line 2 amended
            file 1 line 3
            """).lstrip()
        )
        listing.commit_ref = 'ch17l021'

        with self.assertRaises(ApplyCommitException):
            self.sourcetree.apply_listing_from_commit(listing)


    def _checkout_commit(self, commit):
        commit_spec = self.sourcetree.get_commit_spec(commit)
        self.sourcetree.run_command('git checkout ' + commit_spec)
        self.sourcetree.run_command('git reset')


    def test_raises_if_too_many_files_in_commit(self):
        listing = CodeListing(filename='file1.txt', contents=dedent(
            """
            file 1 line 1
            file 1 line 2
            """).lstrip()
        )
        listing.commit_ref = 'ch17l023'

        self._checkout_commit('ch17l022')
        with self.assertRaises(ApplyCommitException):
            self.sourcetree.apply_listing_from_commit(listing)


    def test_raises_if_listing_doesnt_show_all_new_lines_in_diff(self):
        listing = CodeListing(filename='file1.txt', contents=dedent(
            """
            file 1 line 3
            """).lstrip()
        )
        listing.commit_ref = 'ch17l021'

        with self.assertRaises(ApplyCommitException):
            self.sourcetree.apply_listing_from_commit(listing)


    def test_raises_if_listing_lines_in_wrong_order(self):
        listing = CodeListing(filename='file1.txt', contents=dedent(
            """
            file 1 line 3
#.........这里部分代码省略.........