def test_revoke_subkey(self, pkspec, skspec):
        if pkspec not in self.keys:
            pytest.skip('Keyspec {} not in keys; must not have generated'.format(pkspec))

        alg, size = skspec
        if not alg.can_gen:
            pytest.xfail('Key algorithm {} not yet supported'.format(alg.name))

        if isinstance(size, EllipticCurveOID) and ((not size.can_gen) or size.name not in _openssl_get_supported_curves()):
            pytest.xfail('Curve {} not yet supported'.format(size.name))

        # revoke the subkey
        key = self.keys[pkspec]
        # pub = key.pubkey

        subkey = next(sk for si, sk in key.subkeys.items() if (sk.key_algorithm, sk.key_size) == skspec)

        with key.unlock('This Password Has Been Changed') as ukey:
            rsig = ukey.revoke(subkey, sigtype=SignatureType.SubkeyRevocation)

        assert 'ReasonForRevocation' in rsig._signature.subpackets

        subkey |= rsig

        # verify with PGPy
        assert key.verify(subkey, rsig)

        # try to verify with GPG
        self.gpg_verify_key(key)

def check_xfail_no_run(item):
    """check xfail(run=False)"""
    if not item.config.option.runxfail:
        evalxfail = item._evalxfail
        if evalxfail.istrue():
            if not evalxfail.get('run', True):
                pytest.xfail("[NOTRUN] " + evalxfail.getexplanation())

    def test_that_user_can_purchase_an_app(self, base_url, selenium, new_user):
        if '-dev' not in base_url:
            pytest.skip("Payments can only be tested on dev.")
        else:
            pytest.xfail("Bug 1212152 - App purchases are failing on dev")

        home_page = Home(base_url, selenium)
        home_page.go_to_homepage()
        home_page.header.click_sign_in()
        home_page.login(new_user['email'], new_user['password'])
        assert home_page.is_the_current_page
        home_page.set_region('us')

        # Use the first paid app
        app = home_page.header.search(':paid').results[0]
        app_name = app.name
        details_page = app.click_name()
        assert 'free' not in details_page.price_text
        assert 'paid' in details_page.app_status

        payment = details_page.click_install_button()
        payment.create_pin(self.PIN)
        payment.wait_for_buy_app_section_displayed()
        assert app_name == payment.app_name

        payment.click_buy_button()
        # We are not able to interact with the doorhanger that appears to install the app
        # using Selenium
        # We can check for the `purchased` attribute on the price button though
        details_page.wait_for_app_purchased()

    def test_create_from_format(self):

        pytest.xfail("check static function implementation")

        output = self.run('''
            $tz = new DateTimeZone('America/New_York');
            $format = 'Y-m-d H:i:s';
            $date = DateTime::createFromFormat($format, '2009-02-15 15:16:17', $tz);
            echo $date->format('Y-m-d H:i:s');
        ''')

        assert self.space.str_w(output.pop(0)) == '2009-02-15 15:16:17'

        output = self.run('''
            $tz = new DateTimeZone('America/New_York');
            $format = 'Y-m-!d H:i:s';
            $date = DateTime::createFromFormat($format, '2009-02-15 15:16:17', $tz);
            echo $date->format('Y-m-d H:i:s');
        ''')

        assert self.space.str_w(output.pop(0)) == '1970-01-15 15:16:17'

        output = self.run('''
            $tz = new DateTimeZone('America/New_York');
            $format = '!d';
            $date = DateTime::createFromFormat($format, '15', $tz);
            echo $date->format('Y-m-d H:i:s');
        ''')

        assert self.space.str_w(output.pop(0)) == '1970-01-15 00:00:00'

 def test_sort(self, vector):
   if vector.get_value('table_format').file_format == 'hbase':
     pytest.xfail(reason="IMPALA-283 - select count(*) produces inconsistent results")
   vector.get_value('exec_option')['disable_outermost_topn'] = 1
   self.run_test_case('QueryTest/sort', vector)
   # We can get the sort tests for free from the top-n file
   self.run_test_case('QueryTest/top-n', vector)

def test_pyspark_roundtrip(tempdir, scheme, row_groups, comp, sql):
    if comp in ['BROTLI', 'ZSTD', 'LZO', "LZ4"]:
        pytest.xfail("spark doesn't support compression")
    data = pd.DataFrame({'i32': np.random.randint(-2**17, 2**17, size=1001,
                                                  dtype=np.int32),
                         'i64': np.random.randint(-2**33, 2**33, size=1001,
                                                  dtype=np.int64),
                         'f': np.random.randn(1001),
                         'bhello': np.random.choice([b'hello', b'you',
                            b'people'], size=1001).astype("O"),
                         't': [datetime.datetime.now()]*1001})

    data['t'] += pd.to_timedelta('1ns')
    data['hello'] = data.bhello.str.decode('utf8')
    data.loc[100, 'f'] = np.nan
    data['bcat'] = data.bhello.astype('category')
    data['cat'] = data.hello.astype('category')

    fname = os.path.join(tempdir, 'test.parquet')
    write(fname, data, file_scheme=scheme, row_group_offsets=row_groups,
          compression=comp, times='int96', write_index=True)

    df = sql.read.parquet(fname)
    ddf = df.sort('index').toPandas()
    for col in data:
        if data[col].dtype.kind == "M":
            # pyspark auto-converts timezones
            offset = round((datetime.datetime.utcnow() -
                            datetime.datetime.now()).seconds / 3600)
            ddf[col] + datetime.timedelta(hours=offset) == data[col]
        else:
            assert (ddf[col] == data[col])[~ddf[col].isnull()].all()

def test_fits_mixins_per_column(table_cls, name_col, tmpdir):
    """Test write/read one col at a time and do detailed validation"""
    filename = str(tmpdir.join('test_simple.fits'))
    name, col = name_col

    c = [1.0, 2.0]
    t = table_cls([c, col, c], names=['c1', name, 'c2'])
    t[name].info.description = 'my \n\n\n description'
    t[name].info.meta = {'list': list(range(50)), 'dict': {'a': 'b' * 200}}

    if not t.has_mixin_columns:
        pytest.skip('column is not a mixin (e.g. Quantity subclass in Table)')

    if isinstance(t[name], NdarrayMixin):
        pytest.xfail('NdarrayMixin not supported')

    t.write(filename, format="fits")
    t2 = table_cls.read(filename, format='fits', astropy_native=True)

    assert t.colnames == t2.colnames

    for colname in t.colnames:
        assert_objects_equal(t[colname], t2[colname], compare_attrs[colname])

    # Special case to make sure Column type doesn't leak into Time class data
    if name.startswith('tm'):
        assert t2[name]._time.jd1.__class__ is np.ndarray
        assert t2[name]._time.jd2.__class__ is np.ndarray

def test_record_to_stream(camera, previewing, resolution, format_options):
    format, options = format_options
    if resolution == (2592, 1944) and 'resize' not in options:
        pytest.xfail('Cannot encode video at max resolution')
    if resolution == (1920, 1080) and format == 'mjpeg':
        pytest.xfail('Locks up camera')
    stream1 = tempfile.SpooledTemporaryFile()
    stream2 = tempfile.SpooledTemporaryFile()
    camera.start_recording(stream1, format, **options)
    try:
        camera.wait_recording(1)
        verify2 = (
                format != 'h264' or (
                    options.get('inline_headers', True) and
                    options.get('bitrate', 1)
                    )
                )
        if verify2:
            camera.split_recording(stream2)
            camera.wait_recording(1)
        else:
            with pytest.raises(picamera.PiCameraRuntimeError):
                camera.split_recording(stream2)
    finally:
        camera.stop_recording()
    stream1.seek(0)
    if 'resize' in options:
        resolution = options['resize']
    verify_video(stream1, format, resolution)
    if verify2:
        stream2.seek(0)
        verify_video(stream2, format, resolution)

def test_circular_record(camera, previewing, resolution):
    if resolution == (2592, 1944):
        pytest.xfail('Cannot encode video at max resolution')
    stream = picamera.PiCameraCircularIO(camera, seconds=4)
    camera.start_recording(stream, format='h264')
    try:
        # Keep recording until the stream is definitely full and starts
        # removing earlier bits, or until 20 seconds
        start = time.time()
        while stream._length < stream._size and time.time() - start < 20:
            camera.wait_recording(1)
        # Record one more second, then test the result
        camera.wait_recording(1)
    finally:
        camera.stop_recording()
    temp = tempfile.SpooledTemporaryFile()
    for frame in stream.frames:
        if frame.header:
            stream.seek(frame.position)
            break
    while True:
        buf = stream.read1()
        if not buf:
            break
        temp.write(buf)
    temp.seek(0)
    verify_video(temp, 'h264', resolution)

    def test_masked_masked(self, operation_table_type):
        self._setup(operation_table_type)
        """Two masked tables"""
        if operation_table_type is QTable:
            pytest.xfail('Quantity columns do not support masking.')
        t1 = self.t1
        t1m = operation_table_type(self.t1, masked=True)
        t2 = self.t2
        t2m = operation_table_type(self.t2, masked=True)

        # Result should be masked even though not req'd by inner join
        t1m2m = table.join(t1m, t2m, join_type='inner')
        assert t1m2m.masked is True

        # Result should match non-masked result
        t12 = table.join(t1, t2)
        assert np.all(t12.as_array() == np.array(t1m2m))

        # Mask out some values in both tables and make sure they propagate
        t1m['b'].mask[1] = True
        t1m['c'].mask[2] = True
        t2m['d'].mask[2] = True
        t1m2m = table.join(t1m, t2m, join_type='inner', keys='a')
        assert sort_eq(t1m2m.pformat(), [' a  b_1  c  b_2  d ',
                                         '--- --- --- --- ---',
                                         '  1  --  L2 foo  R1',
                                         '  1  --  L2 foo  R2',
                                         '  1 bar  -- foo  R1',
                                         '  1 bar  -- foo  R2',
                                         '  2 bar  L4 bar  --'])

def test_aggregate_normal(resample_method):
    """Check TimeGrouper's aggregation is identical as normal groupby."""

    if resample_method == 'ohlc':
        pytest.xfail(reason='DataError: No numeric types to aggregate')

    data = np.random.randn(20, 4)
    normal_df = DataFrame(data, columns=['A', 'B', 'C', 'D'])
    normal_df['key'] = [1, 2, 3, 4, 5] * 4

    dt_df = DataFrame(data, columns=['A', 'B', 'C', 'D'])
    dt_df['key'] = [datetime(2013, 1, 1), datetime(2013, 1, 2),
                    datetime(2013, 1, 3), datetime(2013, 1, 4),
                    datetime(2013, 1, 5)] * 4

    normal_grouped = normal_df.groupby('key')
    dt_grouped = dt_df.groupby(TimeGrouper(key='key', freq='D'))

    expected = getattr(normal_grouped, resample_method)()
    dt_result = getattr(dt_grouped, resample_method)()
    expected.index = date_range(start='2013-01-01', freq='D',
                                periods=5, name='key')
    tm.assert_equal(expected, dt_result)

    # if TimeGrouper is used included, 'nth' doesn't work yet

    """

def check_open_tabs(quteproc, request, tabs):
    """Check the list of open tabs in the session.

    This is a lightweight alternative for "The session should look like: ...".

    It expects a list of URLs, with an optional "(active)" suffix.
    """
    if request.config.getoption('--qute-bdd-webengine'):
        pytest.xfail(reason="QtWebEngine TODO: Sessions are not implemented")
    session = quteproc.get_session()
    active_suffix = ' (active)'
    tabs = tabs.splitlines()
    assert len(session['windows']) == 1
    assert len(session['windows'][0]['tabs']) == len(tabs)

    for i, line in enumerate(tabs):
        line = line.strip()
        assert line.startswith('- ')
        line = line[2:]  # remove "- " prefix
        if line.endswith(active_suffix):
            path = line[:-len(active_suffix)]
            active = True
        else:
            path = line
            active = False

        session_tab = session['windows'][0]['tabs'][i]
        assert session_tab['history'][-1]['url'] == quteproc.path_to_url(path)
        if active:
            assert session_tab['active']
        else:
            assert 'active' not in session_tab

    def test_include_na(self, sparse, dtype):
        if sparse:
            pytest.xfail(reason='nan in index is problematic (GH 16894)')

        s = ['a', 'b', np.nan]
        res = get_dummies(s, sparse=sparse, dtype=dtype)
        exp = DataFrame({'a': [1, 0, 0],
                         'b': [0, 1, 0]},
                        dtype=self.effective_dtype(dtype))
        assert_frame_equal(res, exp)

        # Sparse dataframes do not allow nan labelled columns, see #GH8822
        res_na = get_dummies(s, dummy_na=True, sparse=sparse, dtype=dtype)
        exp_na = DataFrame({nan: [0, 0, 1],
                            'a': [1, 0, 0],
                            'b': [0, 1, 0]},
                           dtype=self.effective_dtype(dtype))
        exp_na = exp_na.reindex(['a', 'b', nan], axis=1)
        # hack (NaN handling in assert_index_equal)
        exp_na.columns = res_na.columns
        assert_frame_equal(res_na, exp_na)

        res_just_na = get_dummies([nan], dummy_na=True,
                                  sparse=sparse, dtype=dtype)
        exp_just_na = DataFrame(Series(1, index=[0]), columns=[nan],
                                dtype=self.effective_dtype(dtype))
        tm.assert_numpy_array_equal(res_just_na.values, exp_just_na.values)

def test_ecsv_mixins_per_column(table_cls, name_col):
    """Test write/read one col at a time and do detailed validation"""
    name, col = name_col

    c = [1.0, 2.0]
    t = table_cls([c, col, c], names=['c1', name, 'c2'])
    t[name].info.description = 'description'

    if not t.has_mixin_columns:
        pytest.skip('column is not a mixin (e.g. Quantity subclass in Table)')

    if isinstance(t[name], NdarrayMixin):
        pytest.xfail('NdarrayMixin not supported')

    out = StringIO()
    t.write(out, format="ascii.ecsv")
    t2 = table_cls.read(out.getvalue(), format='ascii.ecsv')

    assert t.colnames == t2.colnames

    for colname in t.colnames:
        assert_objects_equal(t[colname], t2[colname], compare_attrs[colname])

    # Special case to make sure Column type doesn't leak into Time class data
    if name.startswith('tm'):
        assert t2[name]._time.jd1.__class__ is np.ndarray
        assert t2[name]._time.jd2.__class__ is np.ndarray

def test_record_to_file(camera, previewing, resolution, filenames_format_options):
    filename1, filename2, format, options = filenames_format_options
    if resolution == (2592, 1944) and 'resize' not in options:
        pytest.xfail('Cannot encode video at max resolution')
    if resolution == (1920, 1080) and format == 'mjpeg':
        pytest.xfail('Locks up camera')
    camera.start_recording(filename1, **options)
    try:
        camera.wait_recording(1)
        verify2 = (
                format != 'h264' or (
                    options.get('inline_headers', True) and
                    options.get('bitrate', 1)
                    )
                )
        if verify2:
            camera.split_recording(filename2)
            camera.wait_recording(1)
        else:
            with pytest.raises(picamera.PiCameraRuntimeError):
                camera.split_recording(filename2)
    finally:
        camera.stop_recording()
    if 'resize' in options:
        resolution = options['resize']
    verify_video(filename1, format, resolution)
    if verify2:
        verify_video(filename2, format, resolution)

def test_rotest():
    rot = None

    def quit():
        with pytest.raises(SystemExit):
            sys.exit()

    def clicked():
        rot.insert(TKNTR.END, "\nClicked at " + time.asctime(), force=True)
        rot.see(TKNTR.END)

    # make our test window
    try:
        top = TKNTR.Tk()
    except Exception as e:
        pytest.xfail(str(e))  # Travis does not have interactive session
    f = TKNTR.Frame(top)

    sc = TKNTR.Scrollbar(f)
    sc.pack(side=TKNTR.RIGHT, fill=TKNTR.Y)
    rot = ROText(f, wrap=TKNTR.WORD, height=10, yscrollcommand=sc.set,
                 focusBackTo=top)
    rot.pack(side=TKNTR.TOP, fill=TKNTR.X, expand=True)
    sc.config(command=rot.yview)
    f.pack(side=TKNTR.TOP, fill=TKNTR.X)

    b = TKNTR.Button(top, text='Click Me', command=clicked)
    b.pack(side=TKNTR.TOP, fill=TKNTR.X, expand=1)

    q = TKNTR.Button(top, text='Quit', command=quit)
    q.pack(side=TKNTR.TOP)

    # start
    top.mainloop()

 def test_cancel_insert(self, vector):
   self.execute_cancel_test(vector)
   metric_verifier = MetricVerifier(self.impalad_test_service)
   try:
     metric_verifier.verify_no_open_files(timeout=10)
   except AssertionError:
     pytest.xfail("IMPALA-551: File handle leak for INSERT")

def test_missing_residues_xtal_2jaj(request, mol):
    if mol == 'pdb_2jaj_roundtrip':
        pytest.xfail('Writing missing residue records is not yet supported.')
    mol = request.getfixturevalue(mol)
    missingres = mol.metadata.missing_residues
    for expected in MISSINGRES_2JAJ:
        assert missingres[expected[0]][expected[2]] == expected[1]

    def qerun(self, command, stdin_text=None, exp_returncode=0, exp_output=None):
        """
        qerun :: <command> [stdin_text=<string to pass as stdin>] 
            [exp_returncode=<retcode>]
            [<exp_output=<string to check from output>]
        - function to run a command and check return code and output

	:param str command: Command 
	:param str stdin_text: Stdin
	:param int exp_returncode: Return code (default 0)
	:param str exp_output: Check the expected output
        """
        cmd = self.run_command(command, stdin_text, raiseonerr=False)
        if cmd.returncode != exp_returncode:
            pytest.xfail("returncode mismatch.")
            print("GOT: ", cmd.returncode)
            print("EXPECTED: ", exp_returncode)

        if exp_output == None:
            print("Not checking expected output")

        elif cmd.stdout_text.find(exp_output) == 0:
            pytest.xfail("expected output not found")
            print("GOT: ", cmd.stdout_text)
            print("EXPECTED: ", exp_output)

        print("COMMAND SUCCEEDED!")

def test_k_means_fit_predict(algo, dtype, constructor, seed, max_iter, tol):
    # check that fit.predict gives same result as fit_predict
    # There's a very small chance of failure with elkan on unstructured dataset
    # because predict method uses fast euclidean distances computation which
    # may cause small numerical instabilities.
    # NB: This test is largely redundant with respect to test_predict and
    #     test_predict_equal_labels.  This test has the added effect of
    #     testing idempotence of the fittng procesdure which appears to
    #     be where it fails on some MacOS setups.
    if sys.platform == "darwin":
        pytest.xfail(
            "Known failures on MacOS, See "
            "https://github.com/scikit-learn/scikit-learn/issues/12644")
    if not (algo == 'elkan' and constructor is sp.csr_matrix):
        rng = np.random.RandomState(seed)

        X = make_blobs(n_samples=1000, n_features=10, centers=10,
                       random_state=rng)[0].astype(dtype, copy=False)
        X = constructor(X)

        kmeans = KMeans(algorithm=algo, n_clusters=10, random_state=seed,
                        tol=tol, max_iter=max_iter, n_jobs=1)

        labels_1 = kmeans.fit(X).predict(X)
        labels_2 = kmeans.fit_predict(X)

        assert_array_equal(labels_1, labels_2)